openentropy_core/

pool.rs

//! Multi-source entropy pool with health monitoring.
//!
//! Architecture:
//! 1. Auto-discover available sources on this machine
//! 2. Collect raw entropy from each source in parallel
//! 3. Concatenate source bytes into a shared buffer
//! 4. Apply conditioning (Raw / VonNeumann / SHA-256) on output
//! 5. Continuous health monitoring per source
//! 6. Graceful degradation when sources fail
//! 7. Thread-safe for concurrent access

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use sha2::{Digest, Sha256};

use crate::conditioning::{quick_autocorrelation_lag1, quick_min_entropy, quick_shannon};
use crate::source::{EntropySource, SourceState};

const SOURCE_TIMEOUT_BACKOFF_SECS: u64 = 30;

/// Thread-safe multi-source entropy pool.
pub struct EntropyPool {
    sources: Vec<Arc<Mutex<SourceState>>>,
    buffer: Mutex<Vec<u8>>,
    state: Mutex<[u8; 32]>,
    counter: Mutex<u64>,
    total_output: Mutex<u64>,
    // Per-source collection coordination for timeout-safe parallel collection.
    in_flight: Arc<Mutex<HashSet<usize>>>,
    backoff_until: Arc<Mutex<HashMap<usize, Instant>>>,
}

impl EntropyPool {
    /// Create an empty pool.
    pub fn new(seed: Option<&[u8]>) -> Self {
        let initial_state = {
            let mut h = Sha256::new();
            if let Some(s) = seed {
                h.update(s);
            } else {
                // Use OS entropy for initial state
                let mut os_random = [0u8; 32];
                getrandom(&mut os_random);
                h.update(os_random);
            }
            let digest: [u8; 32] = h.finalize().into();
            digest
        };

        Self {
            sources: Vec::new(),
            buffer: Mutex::new(Vec::new()),
            state: Mutex::new(initial_state),
            counter: Mutex::new(0),
            total_output: Mutex::new(0),
            in_flight: Arc::new(Mutex::new(HashSet::new())),
            backoff_until: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Create a pool with all available sources on this machine.
    pub fn auto() -> Self {
        let mut pool = Self::new(None);
        for source in crate::platform::detect_available_sources() {
            pool.add_source(source);
        }
        pool
    }

    /// Register an entropy source.
    pub fn add_source(&mut self, source: Box<dyn EntropySource>) {
        self.sources
            .push(Arc::new(Mutex::new(SourceState::new(source))));
    }

    /// Number of registered sources.
    pub fn source_count(&self) -> usize {
        self.sources.len()
    }

    /// Collect entropy from every registered source in parallel.
    ///
    /// Uses a 10s collection timeout per cycle. Slow sources are skipped and
    /// temporarily backed off to keep callers responsive.
    pub fn collect_all(&self) -> usize {
        self.collect_all_parallel_n(10.0, 1000)
    }

    /// Collect entropy from all sources in parallel using detached worker threads.
    ///
    /// Slow or hung sources are skipped after `timeout_secs`. Timed-out sources
    /// enter a backoff window to avoid thread buildup on repeated calls.
    pub fn collect_all_parallel(&self, timeout_secs: f64) -> usize {
        self.collect_all_parallel_n(timeout_secs, 1000)
    }

    /// Collect entropy from all sources in parallel using detached worker threads.
    ///
    /// - `timeout_secs`: max wall-clock time to wait for a collection cycle.
    /// - `n_samples`: samples requested from each source in this cycle.
    ///
    /// Slow or hung sources are skipped after `timeout_secs`. Timed-out sources
    /// enter a backoff window to avoid thread buildup on repeated calls.
    pub fn collect_all_parallel_n(&self, timeout_secs: f64, n_samples: usize) -> usize {
        let timeout = Duration::from_secs_f64(timeout_secs.max(0.0));
        if timeout.is_zero() || n_samples == 0 {
            return 0;
        }

        let now = Instant::now();
        let mut scheduled: Vec<usize> = Vec::new();
        let mut to_launch: Vec<(usize, Arc<Mutex<SourceState>>)> = Vec::new();

        for (idx, ss_mutex) in self.sources.iter().enumerate() {
            // Skip sources still in backoff.
            let in_backoff = {
                let backoff = self.backoff_until.lock().unwrap();
                backoff.get(&idx).is_some_and(|until| now < *until)
            };
            if in_backoff {
                continue;
            }

            // Skip sources with an in-flight worker from a prior timeout.
            {
                let mut in_flight = self.in_flight.lock().unwrap();
                if in_flight.contains(&idx) {
                    continue;
                }
                in_flight.insert(idx);
            }

            scheduled.push(idx);
            to_launch.push((idx, Arc::clone(ss_mutex)));
        }

        if scheduled.is_empty() {
            return 0;
        }

        // Limit concurrent collection threads to avoid resource exhaustion.
        // Many sources use mmap, JIT pages, socket pairs, large allocations —
        // running all 50+ simultaneously can cause SIGSEGV from memory pressure.
        let max_concurrent = num_cpus().min(16);
        let (tx, rx) = std::sync::mpsc::channel::<(usize, Vec<u8>)>();
        let mut results = Vec::new();
        let mut received = HashSet::new();

        for chunk in to_launch.chunks(max_concurrent) {
            let batch_start = Instant::now();
            let chunk_indices: HashSet<usize> = chunk.iter().map(|&(idx, _)| idx).collect();

            for &(idx, ref src) in chunk {
                let tx = tx.clone();
                let src = Arc::clone(src);
                let in_flight = Arc::clone(&self.in_flight);

                std::thread::spawn(move || {
                    let data = Self::collect_one_n(&src, n_samples);
                    {
                        let mut in_flight = in_flight.lock().unwrap();
                        in_flight.remove(&idx);
                    }
                    let _ = tx.send((idx, data));
                });
            }

            // Wait for this batch to finish. Each batch gets its own full timeout
            // window so that slow sources in early batches don't starve later ones.
            // Only count messages belonging to this chunk's indices — late messages
            // from prior chunks are still collected but don't advance batch_done.
            let mut batch_done = 0;
            while batch_done < chunk.len() {
                let remaining = timeout.saturating_sub(batch_start.elapsed());
                if remaining.is_zero() {
                    break;
                }
                match rx.recv_timeout(remaining) {
                    Ok((idx, data)) => {
                        received.insert(idx);
                        if !data.is_empty() {
                            results.extend_from_slice(&data);
                        }
                        if chunk_indices.contains(&idx) {
                            batch_done += 1;
                        }
                    }
                    Err(_) => break,
                }
            }
        }
        drop(tx);

        // Drain any remaining results from threads that finished after batch loops.
        // Use a brief recv_timeout to catch threads completing just after the
        // batch loops ended, rather than breaking on the first empty try_recv.
        let drain_deadline = Instant::now() + Duration::from_millis(50);
        while received.len() < scheduled.len() {
            let remaining = drain_deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                break;
            }
            match rx.recv_timeout(remaining) {
                Ok((idx, data)) => {
                    received.insert(idx);
                    if !data.is_empty() {
                        results.extend_from_slice(&data);
                    }
                }
                Err(_) => break,
            }
        }

        // Back off any sources that did not respond in time.
        let backoff_for = Duration::from_secs(SOURCE_TIMEOUT_BACKOFF_SECS);
        let timeout_mark = Instant::now() + backoff_for;
        for idx in scheduled {
            if received.contains(&idx) {
                continue;
            }

            {
                let mut bo = self.backoff_until.lock().unwrap();
                bo.insert(idx, timeout_mark);
            }

            if let Ok(mut ss) = self.sources[idx].try_lock() {
                ss.failures += 1;
                ss.healthy = false;
            }
        }

        let n = results.len();
        self.buffer.lock().unwrap().extend_from_slice(&results);
        n
    }

    /// Collect entropy only from sources whose names are in the given list.
    /// Uses parallel threads. Collects 1000 samples per source.
    pub fn collect_enabled(&self, enabled_names: &[String]) -> usize {
        self.collect_enabled_n(enabled_names, 1000)
    }

    /// Collect `n_samples` of entropy from sources whose names are in the list.
    /// Smaller `n_samples` values are faster — use this for interactive/TUI contexts.
    ///
    /// Uses detached threads with a 10-second timeout to avoid blocking
    /// indefinitely on slow or hung sources.
    pub fn collect_enabled_n(&self, enabled_names: &[String], n_samples: usize) -> usize {
        let timeout = Duration::from_secs(10);
        let (tx, rx) = std::sync::mpsc::channel::<(usize, Vec<u8>)>();

        let mut to_launch: Vec<(usize, Arc<Mutex<SourceState>>)> = Vec::new();
        for (idx, ss_mutex) in self.sources.iter().enumerate() {
            let matches = {
                let ss = ss_mutex.lock().unwrap();
                enabled_names.iter().any(|n| n == ss.source.info().name)
            };
            if matches {
                to_launch.push((idx, Arc::clone(ss_mutex)));
            }
        }

        if to_launch.is_empty() {
            return 0;
        }

        let total = to_launch.len();
        let max_concurrent = num_cpus().min(16);
        let mut results = Vec::new();
        let mut received = HashSet::new();

        for chunk in to_launch.chunks(max_concurrent) {
            let batch_start = Instant::now();
            let chunk_indices: HashSet<usize> = chunk.iter().map(|&(idx, _)| idx).collect();

            for &(idx, ref ss_mutex) in chunk {
                let tx = tx.clone();
                let ss_mutex = Arc::clone(ss_mutex);
                std::thread::spawn(move || {
                    let data = Self::collect_one_n(&ss_mutex, n_samples);
                    let _ = tx.send((idx, data));
                });
            }

            let mut batch_done = 0;
            while batch_done < chunk.len() {
                let remaining = timeout.saturating_sub(batch_start.elapsed());
                if remaining.is_zero() {
                    break;
                }
                match rx.recv_timeout(remaining) {
                    Ok((idx, data)) => {
                        received.insert(idx);
                        if !data.is_empty() {
                            results.extend_from_slice(&data);
                        }
                        if chunk_indices.contains(&idx) {
                            batch_done += 1;
                        }
                    }
                    Err(_) => break,
                }
            }
        }
        drop(tx);

        // Brief drain for threads finishing just after batch loops.
        let drain_deadline = Instant::now() + Duration::from_millis(50);
        while received.len() < total {
            let remaining = drain_deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                break;
            }
            match rx.recv_timeout(remaining) {
                Ok((idx, data)) => {
                    received.insert(idx);
                    if !data.is_empty() {
                        results.extend_from_slice(&data);
                    }
                }
                Err(_) => break,
            }
        }

        let n = results.len();
        self.buffer.lock().unwrap().extend_from_slice(&results);
        n
    }

    /// Collect raw bytes from selected sources while preserving source boundaries.
    ///
    /// A single shared timeout budget is applied across the whole sweep, so
    /// callers can bound wall-clock latency even when many slow sources are enabled.
    #[doc(hidden)]
    pub fn collect_enabled_raw_n(
        &self,
        enabled_names: &[String],
        timeout_secs: f64,
        n_samples: usize,
    ) -> HashMap<String, Vec<u8>> {
        let timeout = Duration::from_secs_f64(timeout_secs.max(0.0));
        if timeout.is_zero() || n_samples == 0 {
            return HashMap::new();
        }

        let now = Instant::now();
        let mut to_launch: Vec<(usize, Arc<Mutex<SourceState>>)> = Vec::new();
        let mut source_names = HashMap::new();

        for (idx, ss_mutex) in self.sources.iter().enumerate() {
            let source_name = {
                let ss = ss_mutex.lock().unwrap();
                let name = ss.source.info().name;
                if !enabled_names.iter().any(|enabled| enabled == name) {
                    continue;
                }
                name.to_string()
            };

            let in_backoff = {
                let backoff = self.backoff_until.lock().unwrap();
                backoff.get(&idx).is_some_and(|until| now < *until)
            };
            if in_backoff {
                continue;
            }

            {
                let mut in_flight = self.in_flight.lock().unwrap();
                if in_flight.contains(&idx) {
                    continue;
                }
                in_flight.insert(idx);
            }

            source_names.insert(idx, source_name);
            to_launch.push((idx, Arc::clone(ss_mutex)));
        }

        if to_launch.is_empty() {
            return HashMap::new();
        }

        let max_concurrent = num_cpus().min(16);
        let (tx, rx) = std::sync::mpsc::channel::<(usize, Vec<u8>)>();
        let mut results = HashMap::new();
        let mut received = HashSet::new();
        let mut scheduled = Vec::new();
        let sweep_deadline = Instant::now()
            .checked_add(timeout)
            .unwrap_or_else(Instant::now);

        'batches: for chunk in to_launch.chunks(max_concurrent) {
            if sweep_deadline
                .saturating_duration_since(Instant::now())
                .is_zero()
            {
                break;
            }

            let chunk_indices: HashSet<usize> = chunk.iter().map(|&(idx, _)| idx).collect();
            for &(idx, ref src) in chunk {
                scheduled.push(idx);

                let tx = tx.clone();
                let src = Arc::clone(src);
                let in_flight = Arc::clone(&self.in_flight);

                std::thread::spawn(move || {
                    let data = Self::collect_one_n(&src, n_samples);
                    {
                        let mut in_flight = in_flight.lock().unwrap();
                        in_flight.remove(&idx);
                    }
                    let _ = tx.send((idx, data));
                });
            }

            let mut batch_done = 0;
            while batch_done < chunk.len() {
                let remaining = sweep_deadline.saturating_duration_since(Instant::now());
                if remaining.is_zero() {
                    break 'batches;
                }
                match rx.recv_timeout(remaining) {
                    Ok((idx, data)) => {
                        received.insert(idx);
                        results.insert(idx, data);
                        if chunk_indices.contains(&idx) {
                            batch_done += 1;
                        }
                    }
                    Err(_) => break 'batches,
                }
            }
        }
        drop(tx);

        while received.len() < scheduled.len() {
            let remaining = sweep_deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                break;
            }
            match rx.recv_timeout(remaining.min(Duration::from_millis(50))) {
                Ok((idx, data)) => {
                    received.insert(idx);
                    results.insert(idx, data);
                }
                Err(_) => break,
            }
        }

        let backoff_for = Duration::from_secs(SOURCE_TIMEOUT_BACKOFF_SECS);
        let timeout_mark = Instant::now() + backoff_for;
        for idx in scheduled {
            if received.contains(&idx) {
                continue;
            }

            {
                let mut bo = self.backoff_until.lock().unwrap();
                bo.insert(idx, timeout_mark);
            }

            if let Ok(mut ss) = self.sources[idx].try_lock() {
                ss.failures += 1;
                ss.healthy = false;
            }
        }

        let mut raw_by_source = HashMap::new();
        for (idx, data) in results {
            if data.is_empty() {
                continue;
            }
            if let Some(source_name) = source_names.remove(&idx) {
                raw_by_source.insert(source_name, data);
            }
        }
        raw_by_source
    }

    fn collect_one_n(ss_mutex: &Arc<Mutex<SourceState>>, n_samples: usize) -> Vec<u8> {
        // Clone the Arc<dyn EntropySource> so we can release the mutex during
        // the (potentially slow) collect() call. This allows health_report()
        // and TUI reads to proceed without blocking on source collection.
        let source = {
            let ss = ss_mutex.lock().unwrap();
            Arc::clone(&ss.source)
        };

        let t0 = Instant::now();
        match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| source.collect(n_samples))) {
            Ok(data) if !data.is_empty() => {
                let mut ss = ss_mutex.lock().unwrap();
                ss.last_collect_time = t0.elapsed();
                ss.total_bytes += data.len() as u64;
                ss.last_entropy = quick_shannon(&data);
                ss.last_min_entropy = quick_min_entropy(&data);
                ss.last_autocorrelation = quick_autocorrelation_lag1(&data);
                ss.healthy = ss.last_entropy > 1.0;
                data
            }
            Ok(_) => {
                let mut ss = ss_mutex.lock().unwrap();
                ss.last_collect_time = t0.elapsed();
                ss.failures += 1;
                ss.healthy = false;
                Vec::new()
            }
            Err(_) => {
                let mut ss = ss_mutex.lock().unwrap();
                ss.last_collect_time = t0.elapsed();
                ss.failures += 1;
                ss.healthy = false;
                Vec::new()
            }
        }
    }

    /// Return up to `n_bytes` of raw, unconditioned entropy (XOR-combined only).
    ///
    /// No SHA-256, no DRBG, no whitening. Preserves the raw hardware noise
    /// signal for researchers studying actual device entropy characteristics.
    ///
    /// If sources cannot provide enough bytes after several collection rounds,
    /// this returns the available bytes rather than blocking indefinitely.
    pub fn get_raw_bytes(&self, n_bytes: usize) -> Vec<u8> {
        const MAX_COLLECTION_ROUNDS: usize = 8;

        let mut rounds = 0usize;
        loop {
            let ready = { self.buffer.lock().unwrap().len() >= n_bytes };
            if ready || rounds >= MAX_COLLECTION_ROUNDS {
                break;
            }

            let n = self.collect_all();
            rounds += 1;
            if n == 0 {
                std::thread::sleep(Duration::from_millis(1));
            }
        }

        let mut buf = self.buffer.lock().unwrap();
        let take = n_bytes.min(buf.len());
        if take == 0 {
            return Vec::new();
        }
        let output: Vec<u8> = buf.drain(..take).collect();
        drop(buf);
        *self.total_output.lock().unwrap() += take as u64;
        output
    }

    /// Return `n_bytes` of conditioned random output.
    pub fn get_random_bytes(&self, n_bytes: usize) -> Vec<u8> {
        // Auto-collect if buffer is low
        {
            let buf = self.buffer.lock().unwrap();
            if buf.len() < n_bytes * 2 {
                drop(buf);
                self.collect_all();
            }
        }

        let mut output = Vec::with_capacity(n_bytes);
        while output.len() < n_bytes {
            let mut counter = self.counter.lock().unwrap();
            *counter += 1;
            let cnt = *counter;
            drop(counter);

            // Take up to 256 bytes from buffer
            let sample = {
                let mut buf = self.buffer.lock().unwrap();
                let take = buf.len().min(256);
                let sample: Vec<u8> = buf.drain(..take).collect();
                sample
            };

            // SHA-256 conditioning
            let mut h = Sha256::new();
            let state = self.state.lock().unwrap();
            h.update(*state);
            drop(state);
            h.update(&sample);
            h.update(cnt.to_le_bytes());

            let ts = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default();
            h.update(ts.as_nanos().to_le_bytes());

            // Mix in OS entropy as safety net
            let mut os_random = [0u8; 8];
            getrandom(&mut os_random);
            h.update(os_random);

            let digest: [u8; 32] = h.finalize().into();
            output.extend_from_slice(&digest);

            // Derive state separately from output for forward secrecy.
            let mut sh = Sha256::new();
            sh.update(digest);
            sh.update(b"openentropy_state");
            let new_state: [u8; 32] = sh.finalize().into();
            *self.state.lock().unwrap() = new_state;
        }

        *self.total_output.lock().unwrap() += n_bytes as u64;
        output.truncate(n_bytes);
        output
    }

    /// Return `n_bytes` of entropy with the specified conditioning mode.
    ///
    /// - `Raw`: XOR-combined source bytes, no whitening
    /// - `VonNeumann`: debiased but structure-preserving
    /// - `Sha256`: full cryptographic conditioning (default)
    pub fn get_bytes(
        &self,
        n_bytes: usize,
        mode: crate::conditioning::ConditioningMode,
    ) -> Vec<u8> {
        use crate::conditioning::ConditioningMode;
        match mode {
            ConditioningMode::Raw => self.get_raw_bytes(n_bytes),
            ConditioningMode::VonNeumann => {
                // VN debiasing yields ~25% of input, so collect 6x
                let raw = self.get_raw_bytes(n_bytes * 6);
                crate::conditioning::condition(&raw, n_bytes, ConditioningMode::VonNeumann)
            }
            ConditioningMode::Sha256 => self.get_random_bytes(n_bytes),
        }
    }

    /// Health report as structured data.
    pub fn health_report(&self) -> HealthReport {
        let mut sources = Vec::new();
        let mut healthy_count = 0;
        let mut total_raw = 0u64;

        for ss_mutex in &self.sources {
            let ss = ss_mutex.lock().unwrap();
            if ss.healthy {
                healthy_count += 1;
            }
            total_raw += ss.total_bytes;
            sources.push(SourceHealth {
                name: ss.source.name().to_string(),
                healthy: ss.healthy,
                bytes: ss.total_bytes,
                entropy: ss.last_entropy,
                min_entropy: ss.last_min_entropy,
                autocorrelation: ss.last_autocorrelation,
                time: ss.last_collect_time.as_secs_f64(),
                failures: ss.failures,
            });
        }

        HealthReport {
            healthy: healthy_count,
            total: self.sources.len(),
            raw_bytes: total_raw,
            output_bytes: *self.total_output.lock().unwrap(),
            buffer_size: self.buffer.lock().unwrap().len(),
            sources,
        }
    }

    /// Pretty-print health report.
    pub fn print_health(&self) {
        let r = self.health_report();
        println!("\n{}", "=".repeat(60));
        println!("ENTROPY POOL HEALTH REPORT");
        println!("{}", "=".repeat(60));
        println!("Sources: {}/{} healthy", r.healthy, r.total);
        println!("Raw collected: {} bytes", r.raw_bytes);
        println!(
            "Output: {} bytes | Buffer: {} bytes",
            r.output_bytes, r.buffer_size
        );
        println!(
            "\n{:<25} {:>4} {:>10} {:>6} {:>6} {:>7} {:>5}",
            "Source", "OK", "Bytes", "H", "H∞", "Time", "Fail"
        );
        println!("{}", "-".repeat(68));
        for s in &r.sources {
            let ok = if s.healthy { "✓" } else { "✗" };
            println!(
                "{:<25} {:>4} {:>10} {:>5.2} {:>5.2} {:>6.3}s {:>5}",
                s.name, ok, s.bytes, s.entropy, s.min_entropy, s.time, s.failures
            );
        }
    }

    /// Collect entropy from a single named source and return conditioned bytes.
    ///
    /// Returns `None` if the source name doesn't match any registered source.
    pub fn get_source_bytes(
        &self,
        source_name: &str,
        n_bytes: usize,
        mode: crate::conditioning::ConditioningMode,
    ) -> Option<Vec<u8>> {
        if n_bytes == 0 {
            return Some(Vec::new());
        }

        let ss_mutex = self
            .sources
            .iter()
            .find(|ss_mutex| {
                let ss = ss_mutex.lock().unwrap();
                ss.source.info().name == source_name
            })
            .cloned()?;

        let n_samples = match mode {
            crate::conditioning::ConditioningMode::Raw => n_bytes,
            crate::conditioning::ConditioningMode::VonNeumann => n_bytes * 6,
            crate::conditioning::ConditioningMode::Sha256 => n_bytes * 4 + 64,
        };
        let raw = Self::collect_one_n(&ss_mutex, n_samples);
        if raw.is_empty() {
            return None; // Source failed to produce output
        }
        let output = crate::conditioning::condition(&raw, n_bytes, mode);
        Some(output)
    }

    /// Collect raw bytes from a single named source.
    ///
    /// Returns `None` if no source matches the name.
    pub fn get_source_raw_bytes(&self, source_name: &str, n_samples: usize) -> Option<Vec<u8>> {
        let ss_mutex = self.sources.iter().find(|ss_mutex| {
            let ss = ss_mutex.lock().unwrap();
            ss.source.info().name == source_name
        })?;

        let raw = Self::collect_one_n(ss_mutex, n_samples);
        if raw.is_empty() {
            return None; // Source failed to produce output
        }
        Some(raw)
    }

    /// List all registered source names.
    pub fn source_names(&self) -> Vec<String> {
        self.sources
            .iter()
            .map(|ss_mutex| {
                let ss = ss_mutex.lock().unwrap();
                ss.source.info().name.to_string()
            })
            .collect()
    }

    /// Get source info for each registered source.
    pub fn source_infos(&self) -> Vec<SourceInfoSnapshot> {
        self.sources
            .iter()
            .map(|ss_mutex| {
                let ss = ss_mutex.lock().unwrap();
                let info = ss.source.info();
                SourceInfoSnapshot {
                    name: info.name.to_string(),
                    description: info.description.to_string(),
                    physics: info.physics.to_string(),
                    category: info.category.to_string(),
                    platform: info.platform.to_string(),
                    requirements: info.requirements.iter().map(|r| r.to_string()).collect(),
                    entropy_rate_estimate: info.entropy_rate_estimate,
                    composite: info.composite,
                    config: ss.source.config_options(),
                }
            })
            .collect()
    }

    /// Call a function on a named source, returning `None` if no match.
    pub fn with_source<F, R>(&self, name: &str, f: F) -> Option<R>
    where
        F: FnOnce(&dyn EntropySource) -> R,
    {
        self.sources
            .iter()
            .find(|ss| ss.lock().unwrap().source.info().name == name)
            .map(|ss| f(&*ss.lock().unwrap().source))
    }
}

/// Fill buffer with OS random bytes via the `getrandom` crate.
/// Works cross-platform (Unix, Windows, WASM, etc.) without manual file I/O.
///
/// # Panics
/// Panics if the OS CSPRNG fails — this indicates a fatal platform issue.
fn getrandom(buf: &mut [u8]) {
    getrandom::fill(buf).expect("OS CSPRNG failed");
}

/// Number of logical CPUs (for concurrency limits).
fn num_cpus() -> usize {
    std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(4)
}

/// Overall health report for the entropy pool.
#[derive(Debug, Clone)]
pub struct HealthReport {
    /// Number of healthy sources.
    pub healthy: usize,
    /// Total number of registered sources.
    pub total: usize,
    /// Total raw bytes collected across all sources.
    pub raw_bytes: u64,
    /// Total conditioned output bytes produced.
    pub output_bytes: u64,
    /// Current internal buffer size in bytes.
    pub buffer_size: usize,
    /// Per-source health details.
    pub sources: Vec<SourceHealth>,
}

/// Health status of a single entropy source.
#[derive(Debug, Clone)]
pub struct SourceHealth {
    /// Source name.
    pub name: String,
    /// Whether the source is currently healthy (entropy > 1.0 bits/byte).
    pub healthy: bool,
    /// Total bytes collected from this source.
    pub bytes: u64,
    /// Shannon entropy of the last collection (bits per byte, max 8.0).
    pub entropy: f64,
    /// Min-entropy of the last collection (bits per byte, max 8.0). More conservative than Shannon.
    pub min_entropy: f64,
    /// Lag-1 autocorrelation of the last collection ([-1, 1], 0 = ideal).
    pub autocorrelation: f64,
    /// Time taken for the last collection in seconds.
    pub time: f64,
    /// Number of collection failures.
    pub failures: u64,
}

/// Snapshot of source metadata for external consumption.
#[derive(Debug, Clone)]
pub struct SourceInfoSnapshot {
    /// Source name.
    pub name: String,
    /// Human-readable description.
    pub description: String,
    /// Physics explanation.
    pub physics: String,
    /// Source category.
    pub category: String,
    /// Target platform.
    pub platform: String,
    /// Hardware/software requirements.
    pub requirements: Vec<String>,
    /// Estimated entropy rate.
    pub entropy_rate_estimate: f64,
    /// Whether this is a composite source.
    pub composite: bool,
    /// Runtime configuration keys and current values.
    pub config: Vec<(&'static str, String)>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::source::{Platform, SourceCategory, SourceInfo};

    // -----------------------------------------------------------------------
    // Mock entropy source for testing
    // -----------------------------------------------------------------------

    /// A deterministic mock entropy source that returns predictable data.
    struct MockSource {
        info: SourceInfo,
        data: Vec<u8>,
    }

    impl MockSource {
        fn new(name: &'static str, data: Vec<u8>) -> Self {
            Self {
                info: SourceInfo {
                    name,
                    description: "mock source",
                    physics: "deterministic test data",
                    category: SourceCategory::System,
                    platform: Platform::Any,
                    requirements: &[],
                    entropy_rate_estimate: 1.0,
                    composite: false,
                    is_fast: true,
                },
                data,
            }
        }
    }

    impl EntropySource for MockSource {
        fn info(&self) -> &SourceInfo {
            &self.info
        }
        fn is_available(&self) -> bool {
            true
        }
        fn collect(&self, n_samples: usize) -> Vec<u8> {
            self.data.iter().copied().cycle().take(n_samples).collect()
        }
    }

    /// A mock source that always fails (returns empty).
    struct FailingSource {
        info: SourceInfo,
    }

    impl FailingSource {
        fn new(name: &'static str) -> Self {
            Self {
                info: SourceInfo {
                    name,
                    description: "failing mock",
                    physics: "always fails",
                    category: SourceCategory::System,
                    platform: Platform::Any,
                    requirements: &[],
                    entropy_rate_estimate: 0.0,
                    composite: false,
                    is_fast: true,
                },
            }
        }
    }

    impl EntropySource for FailingSource {
        fn info(&self) -> &SourceInfo {
            &self.info
        }
        fn is_available(&self) -> bool {
            true
        }
        fn collect(&self, _n_samples: usize) -> Vec<u8> {
            Vec::new()
        }
    }

    struct SlowSource {
        info: SourceInfo,
        delay: Duration,
        value: u8,
    }

    impl SlowSource {
        fn new(name: &'static str, delay: Duration, value: u8) -> Self {
            Self {
                info: SourceInfo {
                    name,
                    description: "slow mock",
                    physics: "delayed deterministic test data",
                    category: SourceCategory::System,
                    platform: Platform::Any,
                    requirements: &[],
                    entropy_rate_estimate: 1.0,
                    composite: false,
                    is_fast: false,
                },
                delay,
                value,
            }
        }
    }

    impl EntropySource for SlowSource {
        fn info(&self) -> &SourceInfo {
            &self.info
        }
        fn is_available(&self) -> bool {
            true
        }
        fn collect(&self, n_samples: usize) -> Vec<u8> {
            std::thread::sleep(self.delay);
            vec![self.value; n_samples]
        }
    }

    // -----------------------------------------------------------------------
    // Pool creation tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_pool_new_empty() {
        let pool = EntropyPool::new(None);
        assert_eq!(pool.source_count(), 0);
    }

    #[test]
    fn test_pool_new_with_seed() {
        let pool = EntropyPool::new(Some(b"test seed"));
        assert_eq!(pool.source_count(), 0);
    }

    #[test]
    fn test_pool_add_source() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock1", vec![42])));
        assert_eq!(pool.source_count(), 1);
    }

    #[test]
    fn test_pool_add_multiple_sources() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock1", vec![1])));
        pool.add_source(Box::new(MockSource::new("mock2", vec![2])));
        pool.add_source(Box::new(MockSource::new("mock3", vec![3])));
        assert_eq!(pool.source_count(), 3);
    }

    // -----------------------------------------------------------------------
    // Collection tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_collect_all_returns_bytes() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock1", vec![0xAA, 0xBB, 0xCC])));
        let n = pool.collect_all();
        assert!(n > 0, "Should have collected some bytes");
    }

    #[test]
    fn test_collect_all_parallel_with_timeout() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock1", vec![1, 2])));
        pool.add_source(Box::new(MockSource::new("mock2", vec![3, 4])));
        let n = pool.collect_all_parallel(5.0);
        assert!(n > 0);
    }

    #[test]
    fn test_collect_enabled_filters_sources() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("alpha", vec![1])));
        pool.add_source(Box::new(MockSource::new("beta", vec![2])));

        let enabled = vec!["alpha".to_string()];
        let n = pool.collect_enabled(&enabled);
        assert!(n > 0, "Should collect from enabled source");
    }

    #[test]
    fn test_collect_enabled_no_match() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("alpha", vec![1])));

        let enabled = vec!["nonexistent".to_string()];
        let n = pool.collect_enabled(&enabled);
        assert_eq!(n, 0, "No sources should match");
    }

    #[test]
    fn test_collect_enabled_raw_n_preserves_source_boundaries() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("alpha", vec![1, 2, 3])));
        pool.add_source(Box::new(MockSource::new("beta", vec![9, 8, 7])));

        let enabled = vec!["alpha".to_string(), "beta".to_string()];
        let results = pool.collect_enabled_raw_n(&enabled, 1.0, 4);

        assert_eq!(results.len(), 2);
        assert_eq!(results.get("alpha").unwrap(), &vec![1, 2, 3, 1]);
        assert_eq!(results.get("beta").unwrap(), &vec![9, 8, 7, 9]);
    }

    #[test]
    fn test_collect_enabled_raw_n_respects_shared_timeout_budget() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("fast", vec![1, 2, 3, 4])));
        pool.add_source(Box::new(SlowSource::new(
            "slow",
            Duration::from_millis(500),
            7,
        )));

        let enabled = vec!["fast".to_string(), "slow".to_string()];
        let start = Instant::now();
        let results = pool.collect_enabled_raw_n(&enabled, 0.01, 4);
        let elapsed = start.elapsed();

        assert!(
            elapsed < Duration::from_millis(250),
            "shared timeout budget should return well before the slow source finishes: {elapsed:?}"
        );
        assert_eq!(results.get("fast").unwrap(), &vec![1, 2, 3, 4]);
        assert!(!results.contains_key("slow"));
    }

    #[test]
    fn test_collect_all_parallel_keeps_backoff_after_worker_finishes() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(SlowSource::new(
            "slow",
            Duration::from_millis(500),
            7,
        )));

        assert_eq!(pool.collect_all_parallel_n(0.01, 4), 0);
        std::thread::sleep(Duration::from_millis(600));

        let backoff_until = pool.backoff_until.lock().unwrap().get(&0).copied();
        assert!(backoff_until.is_some());
        assert!(backoff_until.unwrap() > Instant::now());

        let retry_started = Instant::now();
        assert_eq!(pool.collect_all_parallel_n(0.01, 4), 0);
        assert!(retry_started.elapsed() < Duration::from_millis(50));
    }

    #[test]
    fn test_collect_enabled_raw_n_keeps_backoff_after_worker_finishes() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(SlowSource::new(
            "slow",
            Duration::from_millis(500),
            7,
        )));

        let enabled = vec!["slow".to_string()];
        assert!(pool.collect_enabled_raw_n(&enabled, 0.01, 4).is_empty());
        std::thread::sleep(Duration::from_millis(600));

        let backoff_until = pool.backoff_until.lock().unwrap().get(&0).copied();
        assert!(backoff_until.is_some());
        assert!(backoff_until.unwrap() > Instant::now());

        let retry_started = Instant::now();
        assert!(pool.collect_enabled_raw_n(&enabled, 0.01, 4).is_empty());
        assert!(retry_started.elapsed() < Duration::from_millis(50));
    }

    // -----------------------------------------------------------------------
    // Byte output tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_get_raw_bytes_length() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let bytes = pool.get_raw_bytes(64);
        assert_eq!(bytes.len(), 64);
    }

    #[test]
    fn test_get_random_bytes_length() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let bytes = pool.get_random_bytes(64);
        assert_eq!(bytes.len(), 64);
    }

    #[test]
    fn test_get_random_bytes_various_sizes() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        for size in [1, 16, 32, 64, 100, 256] {
            let bytes = pool.get_random_bytes(size);
            assert_eq!(bytes.len(), size, "Expected {size} bytes");
        }
    }

    #[test]
    fn test_get_bytes_raw_mode() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let bytes = pool.get_bytes(32, crate::conditioning::ConditioningMode::Raw);
        assert_eq!(bytes.len(), 32);
    }

    #[test]
    fn test_get_bytes_sha256_mode() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let bytes = pool.get_bytes(32, crate::conditioning::ConditioningMode::Sha256);
        assert_eq!(bytes.len(), 32);
    }

    #[test]
    fn test_get_bytes_von_neumann_mode() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let bytes = pool.get_bytes(16, crate::conditioning::ConditioningMode::VonNeumann);
        // VonNeumann may produce fewer bytes due to debiasing yield
        assert!(bytes.len() <= 16);
    }

    // -----------------------------------------------------------------------
    // Health report tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_health_report_empty_pool() {
        let pool = EntropyPool::new(Some(b"test"));
        let report = pool.health_report();
        assert_eq!(report.total, 0);
        assert_eq!(report.healthy, 0);
        assert_eq!(report.raw_bytes, 0);
        assert_eq!(report.output_bytes, 0);
        assert_eq!(report.buffer_size, 0);
        assert!(report.sources.is_empty());
    }

    #[test]
    fn test_health_report_after_collection() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new(
            "good_source",
            (0..=255).collect(),
        )));
        pool.collect_all();
        let report = pool.health_report();
        assert_eq!(report.total, 1);
        assert!(report.raw_bytes > 0);
        assert_eq!(report.sources.len(), 1);
        assert_eq!(report.sources[0].name, "good_source");
        assert!(report.sources[0].bytes > 0);
    }

    #[test]
    fn test_health_report_failing_source() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(FailingSource::new("bad_source")));
        pool.collect_all();
        let report = pool.health_report();
        assert_eq!(report.total, 1);
        assert_eq!(report.healthy, 0);
        assert!(!report.sources[0].healthy);
        assert_eq!(report.sources[0].failures, 1);
    }

    #[test]
    fn test_health_report_mixed_sources() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("good", (0..=255).collect())));
        pool.add_source(Box::new(FailingSource::new("bad")));
        pool.collect_all();
        let report = pool.health_report();
        assert_eq!(report.total, 2);
        // The good source should be healthy if its entropy > 1.0
        assert!(report.healthy >= 1);
        assert_eq!(report.sources.len(), 2);
    }

    #[test]
    fn test_health_report_tracks_output_bytes() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", (0..=255).collect())));
        let _ = pool.get_random_bytes(64);
        let report = pool.health_report();
        assert!(report.output_bytes >= 64);
    }

    // -----------------------------------------------------------------------
    // Source info snapshot tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_source_infos_empty() {
        let pool = EntropyPool::new(Some(b"test"));
        let infos = pool.source_infos();
        assert!(infos.is_empty());
    }

    #[test]
    fn test_source_infos_populated() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("test_src", vec![1])));
        let infos = pool.source_infos();
        assert_eq!(infos.len(), 1);
        assert_eq!(infos[0].name, "test_src");
        assert_eq!(infos[0].description, "mock source");
        assert_eq!(infos[0].category, "system");
        assert!((infos[0].entropy_rate_estimate - 1.0).abs() < f64::EPSILON);
    }

    // -----------------------------------------------------------------------
    // Determinism / seed tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_different_seeds_differ() {
        let mut pool1 = EntropyPool::new(Some(b"seed_a"));
        pool1.add_source(Box::new(MockSource::new("m", vec![42; 100])));
        let mut pool2 = EntropyPool::new(Some(b"seed_b"));
        pool2.add_source(Box::new(MockSource::new("m", vec![42; 100])));

        let bytes1 = pool1.get_random_bytes(32);
        let bytes2 = pool2.get_random_bytes(32);
        assert_ne!(
            bytes1, bytes2,
            "Different seeds should produce different output"
        );
    }

    // -----------------------------------------------------------------------
    // Edge case tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_collect_from_empty_pool() {
        let pool = EntropyPool::new(Some(b"test"));
        let n = pool.collect_all();
        assert_eq!(n, 0, "Empty pool should collect 0 bytes");
    }

    #[test]
    fn test_collect_enabled_empty_list() {
        let mut pool = EntropyPool::new(Some(b"test"));
        pool.add_source(Box::new(MockSource::new("mock", vec![1])));
        let n = pool.collect_enabled(&[]);
        assert_eq!(n, 0);
    }
}