luff 0.2.1

//! Timeout utilities for safe I/O operations
//!
//! This module provides a safe, cross-platform way to execute blocking operations
//! with a timeout, preventing denial-of-service attacks via blocking syscalls.
//!
//! It uses a static thread pool to execute tasks, preventing thread exhaustion
//! from spawning new threads for every operation.
//!
//! This is intentionally scoped to `fs_utils` — it exists to support
//! `read_file_safe` and similar I/O operations that need timeout protection.
//! If other subsystems need timeout infrastructure, this can be promoted.
//!
//! # Feature gate
//!
//! This module requires the `cli` feature (which pulls in `crossbeam-channel`).
//! It is not available in WASM builds.
#![cfg(feature = "cli")]

use crossbeam_channel::{Receiver, Sender, bounded};
use std::io;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::{OnceLock, mpsc};
use std::thread;
use std::time::Duration;

/// Global default timeout (5 seconds) - can be overridden via `LUFF_READ_TIMEOUT_MS`
const DEFAULT_TIMEOUT_MS: u64 = 5000;

/// A job for the thread pool: a closure that can be sent across threads
type Job = Box<dyn FnOnce() + Send + 'static>;

/// Global thread pool for timeout operations
static POOL: OnceLock<ThreadPool> = OnceLock::new();

/// Thread pool for executing timeout tasks
///
/// Uses a bounded set of worker threads to process blocking operations.
/// Implemented using `crossbeam-channel` for efficient MPMC (Multi-Producer Multi-Consumer)
/// queueing without Mutex contention.
struct ThreadPool {
    /// Channel sender to submit jobs to the workers
    sender: Sender<Job>,
}

impl ThreadPool {
    /// Get the global thread pool instance, initializing it if necessary
    fn get() -> &'static Self {
        POOL.get_or_init(Self::new)
    }

    /// Create a new thread pool
    fn new() -> Self {
        // Use a bounded channel to provide backpressure if the system is overloaded.
        // Capacity is set high enough to avoid blocking under normal load.
        let (sender, receiver): (Sender<Job>, Receiver<Job>) = bounded(4096);

        // Determine pool size based on available CPUs.
        // Since these tasks are I/O bound (file reads), we use a multiplier of the
        // CPU count to ensure we can saturate I/O even if some threads are blocked
        // on syscalls.
        let cpu_count = thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);

        // 4x multiplier for I/O bound tasks, with a reasonable floor and ceiling
        let size = (cpu_count * 4).clamp(4, 64);

        let mut spawned = 0usize;
        for i in 0..size {
            let rx = receiver.clone();
            // Spawn named threads for better debuggability
            let builder = thread::Builder::new().name(format!("luff-io-worker-{i}"));

            match builder.spawn(move || Self::worker_loop(&rx)) {
                Ok(_) => spawned += 1,
                Err(e) => {
                    log::warn!("Failed to spawn IO worker thread {i}: {e}");
                }
            }
        }

        // If we couldn't spawn a single worker, file reads will silently
        // timeout forever. Fail fast instead.
        assert!(
            spawned > 0,
            "Failed to spawn any IO worker threads ({size} attempted)"
        );

        Self { sender }
    }

    /// Worker loop that processes jobs from the channel
    fn worker_loop(receiver: &Receiver<Job>) {
        // Loop until the channel is disconnected (which happens only on process exit
        // since the global sender is static)
        while let Ok(job) = receiver.recv() {
            // Execute job outside any lock to allow full concurrency.
            // Catch panics to prevent the worker thread from dying, ensuring
            // the pool remains stable over time.
            if let Err(cause) = catch_unwind(AssertUnwindSafe(job)) {
                log::error!("IO worker thread caught panic: {cause:?}");
            }
        }
    }

    /// Submit a job to the thread pool
    ///
    /// Uses `try_send` to avoid blocking the caller if the pool is saturated.
    /// Under pathological I/O conditions (e.g., hung NFS mount), all workers
    /// may be blocked on syscalls. With a blocking `send`, the caller's thread
    /// (typically the walker) would also hang indefinitely — the receive-side
    /// timeout would never fire because we'd never reach it. `try_send` turns
    /// this into an immediate, actionable error instead of a silent deadlock.
    ///
    /// # Errors
    ///
    /// Returns an `io::Error` if the channel is full (pool saturated) or
    /// if all worker threads have disconnected.
    fn try_execute<F>(&self, f: F) -> io::Result<()>
    where
        F: FnOnce() + Send + 'static,
    {
        use crossbeam_channel::TrySendError;
        self.sender.try_send(Box::new(f)).map_err(|e| match e {
            TrySendError::Full(_) => io::Error::other(
                "I/O thread pool saturated — all workers are blocked. \
                 This may indicate a hung filesystem (e.g., NFS). \
                 Consider increasing LUFF_READ_TIMEOUT_MS or investigating the mount.",
            ),
            TrySendError::Disconnected(_) => {
                io::Error::other("I/O thread pool workers disconnected unexpectedly")
            }
        })
    }
}

/// Environment provider for timeout configuration
pub trait TimeoutEnv {
    /// Get the timeout duration in milliseconds from environment
    fn timeout_ms(&self) -> Option<u64>;
}

/// Production implementation that reads from process environment
struct RealTimeoutEnv;

impl TimeoutEnv for RealTimeoutEnv {
    fn timeout_ms(&self) -> Option<u64> {
        std::env::var("LUFF_READ_TIMEOUT_MS")
            .ok()
            .and_then(|s| s.parse().ok())
    }
}

/// Get timeout duration with a custom environment provider (for testing)
#[must_use]
fn get_timeout_duration_with_env(env: &dyn TimeoutEnv) -> Duration {
    env.timeout_ms().map_or_else(
        || Duration::from_millis(DEFAULT_TIMEOUT_MS),
        Duration::from_millis,
    )
}

/// Execute a closure with a timeout, returning its result or a timeout error
///
/// The closure can return any `T` — the timeout layer only injects `io::Error`
/// for timeout or thread disconnection. This avoids forcing callers into
/// double-`Result` patterns when their closure has its own error type.
///
/// Uses a shared thread pool to execute the closure, preventing thread exhaustion.
///
/// # Zombie tasks
///
/// When a timeout fires, the closure **continues running** on the worker thread
/// until it completes or errors naturally. A sustained burst of timeouts (e.g.,
/// from a hung NFS mount) can therefore saturate the pool. This is acceptable
/// for `read_file_safe` where kernel file reads are bounded, but callers adding
/// new use-cases should be aware of this constraint.
///
/// # Errors
///
/// Returns an [`io::Error`] if:
/// - The operation times out ([`io::ErrorKind::TimedOut`])
/// - The I/O thread pool is saturated (all workers blocked)
/// - The worker thread panics or disconnects unexpectedly
///
/// The error message will include the configured timeout duration for timeouts.
pub fn with_timeout<F, T>(f: F) -> io::Result<T>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,
{
    TimeoutExecutor::new().execute(f)
}

/// Timeout executor that can use different environment sources
struct TimeoutExecutor {
    /// Environment provider for timeout configuration
    env: Box<dyn TimeoutEnv>,
}

impl TimeoutExecutor {
    /// Create a new executor with the production environment
    fn new() -> Self {
        Self {
            env: Box::new(RealTimeoutEnv),
        }
    }

    /// Create a new executor with a custom environment (for testing)
    #[cfg(test)]
    fn with_env(env: impl TimeoutEnv + 'static) -> Self {
        Self { env: Box::new(env) }
    }

    /// Execute a closure with the configured timeout
    ///
    /// # Arguments
    ///
    /// * `f` - Closure to execute with timeout protection
    ///
    /// # Returns
    ///
    /// Returns `Ok(T)` with the closure's return value, or an `io::Error`
    /// if the operation timed out or the worker disconnected.
    fn execute<F, T>(&self, f: F) -> io::Result<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static,
    {
        let timeout = get_timeout_duration_with_env(&*self.env);
        let (tx, rx) = mpsc::channel();

        // Submit job to global thread pool instead of spawning new thread.
        // try_execute returns immediately with an error if the pool is
        // saturated, preventing the caller from blocking indefinitely.
        ThreadPool::get().try_execute(move || {
            let _ = tx.send(f());
        })?;

        match rx.recv_timeout(timeout) {
            Ok(result) => Ok(result),
            Err(mpsc::RecvTimeoutError::Timeout) => Err(io::Error::new(
                io::ErrorKind::TimedOut,
                format!(
                    "Operation timed out after {}ms (LUFF_READ_TIMEOUT_MS)",
                    timeout.as_millis()
                ),
            )),
            Err(mpsc::RecvTimeoutError::Disconnected) => Err(io::Error::other(
                "Operation thread disconnected unexpectedly",
            )),
        }
    }
}

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

    /// Test environment that returns a fixed timeout value
    struct TestEnv {
        /// Fixed timeout in milliseconds, or None for default
        value: Option<u64>,
    }

    impl TimeoutEnv for TestEnv {
        fn timeout_ms(&self) -> Option<u64> {
            self.value
        }
    }

    #[test]
    fn test_timeout_with_fast_operation() {
        let result = with_timeout(|| "success".to_string());
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "success");
    }

    #[test]
    fn test_timeout_with_send_bound() {
        // Verify Send requirement is enforced
        fn assert_send<T: Send>(_: &T) {}
        let result = with_timeout(|| 42i32);
        assert_send(&result);
    }

    #[test]
    fn test_timeout_with_slow_operation() {
        // This test verifies that operations complete normally within timeout
        let executor = TimeoutExecutor::with_env(TestEnv { value: Some(5000) });
        let result: io::Result<()> = executor.execute(|| {
            sleep(Duration::from_millis(100));
        });
        assert!(result.is_ok());
    }

    #[test]
    fn test_timeout_triggers() {
        let executor = TimeoutExecutor::with_env(TestEnv { value: Some(10) });
        let result: io::Result<()> = executor.execute(|| {
            sleep(Duration::from_millis(100));
        });

        let err = result.expect_err("should time out with 10ms timeout");
        assert_eq!(
            err.kind(),
            io::ErrorKind::TimedOut,
            "timeout should produce TimedOut error kind, got: {err}"
        );
        assert!(
            err.to_string().contains("LUFF_READ_TIMEOUT_MS"),
            "timeout error should mention the env var for discoverability, got: {err}"
        );
    }

    #[test]
    fn test_get_timeout_duration_default() {
        let env = TestEnv { value: None };
        let duration = get_timeout_duration_with_env(&env);
        assert_eq!(duration, Duration::from_secs(5));
    }

    #[test]
    fn test_get_timeout_duration_custom() {
        let env = TestEnv { value: Some(1000) };
        let duration = get_timeout_duration_with_env(&env);
        assert_eq!(duration, Duration::from_secs(1));
    }

    #[test]
    fn test_thread_pool_reuse() {
        // Verify that multiple calls reuse the pool
        let executor = TimeoutExecutor::with_env(TestEnv { value: Some(5000) });

        let mut handles = vec![];
        for i in 0..10 {
            let result = executor.execute(move || i);
            handles.push(result);
        }

        for (i, res) in handles.into_iter().enumerate() {
            assert_eq!(
                res.expect("pool reuse should not fail"),
                i,
                "closure should return its captured value"
            );
        }
    }

    #[test]
    fn test_closure_error_passes_through() {
        // Verify that a closure returning its own Result type passes through
        // without being conflated with timeout errors
        let outer = with_timeout(|| -> Result<(), String> { Err("application error".to_string()) });
        // Outer Result is Ok (no timeout), inner Result is Err (application)
        let inner = outer.expect("timeout layer should succeed");
        assert_eq!(
            inner.unwrap_err(),
            "application error",
            "inner application error should pass through unchanged"
        );
    }

    #[test]
    fn test_timeout_returns_value_types() {
        // Verify various return types work through the Send + 'static bounds
        let int = with_timeout(|| 42u64).expect("u64 return should work");
        assert_eq!(int, 42);

        let vec = with_timeout(|| vec![1, 2, 3]).expect("Vec return should work");
        assert_eq!(vec, [1, 2, 3]);

        let opt = with_timeout(|| Option::<String>::None).expect("Option return should work");
        assert_eq!(opt, None);
    }

    #[test]
    fn test_timeout_zero_ms_triggers_immediately() {
        // A 0ms timeout should fire even for trivial closures (or succeed if
        // the worker is fast enough). Either outcome is acceptable — the key
        // invariant is no panic and no hang.
        let executor = TimeoutExecutor::with_env(TestEnv { value: Some(0) });
        let result: io::Result<()> = executor.execute(|| {
            sleep(Duration::from_millis(50));
        });
        // With 0ms timeout and a 50ms sleep, this should almost always time out,
        // but we only assert it doesn't panic/hang.
        if let Err(e) = result {
            assert_eq!(e.kind(), io::ErrorKind::TimedOut);
        }
    }
}