armdb 0.4.1

//! Background periodic-flush worker.
//!
//! Mirrors [`Compactor`](crate::compaction::Compactor): a thread that wakes on a
//! fixed interval and runs a flush closure, wired to the shared
//! [`ShutdownSignal`](crate::shutdown::ShutdownSignal) so it stops together with
//! the rest of the database's background workers. Unlike compaction, a failing
//! flush is logged and retried on the next tick rather than ending the loop —
//! a missed flush degrades durability but is not a reason to kill the worker.

use std::time::Duration;

use crate::error::DbResult;

/// Background periodic-flush handle. Stops and joins its thread on `stop()` or
/// `Drop`.
pub struct Flusher {
    stop: crate::shutdown::ShutdownSignal,
    handle: Option<std::thread::JoinHandle<()>>,
}

impl Flusher {
    /// Start a background flush thread with its own shutdown signal.
    pub fn start(flush_fn: impl Fn() -> DbResult<()> + Send + 'static, interval: Duration) -> Self {
        Self::start_with_signal(flush_fn, interval, crate::shutdown::ShutdownSignal::new())
    }

    /// Start a background flush thread controlled by an external shutdown
    /// signal. When the signal fires the thread wakes immediately instead of
    /// waiting for the full interval.
    pub fn start_with_signal(
        flush_fn: impl Fn() -> DbResult<()> + Send + 'static,
        interval: Duration,
        signal: crate::shutdown::ShutdownSignal,
    ) -> Self {
        let stop = signal.clone();
        let handle = std::thread::spawn(move || {
            while !stop.is_shutdown() {
                if stop.wait_timeout(interval) {
                    break;
                }
                if let Err(e) = flush_fn() {
                    tracing::error!(error = %e, "flush error");
                }
            }
        });
        Self {
            stop: signal,
            handle: Some(handle),
        }
    }

    /// Signal the worker to stop and join its thread.
    pub fn stop(&mut self) {
        self.stop.shutdown();
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
    }
}

impl Drop for Flusher {
    fn drop(&mut self) {
        self.stop();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn wait_until(pred: impl Fn() -> bool, timeout: Duration) -> bool {
        let start = std::time::Instant::now();
        while start.elapsed() < timeout {
            if pred() {
                return true;
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        pred()
    }

    #[test]
    fn flusher_ticks_and_stop_joins() {
        let n = Arc::new(AtomicUsize::new(0));
        let n2 = n.clone();
        let mut f = Flusher::start(
            move || {
                n2.fetch_add(1, Ordering::SeqCst);
                Ok(())
            },
            Duration::from_millis(10),
        );
        assert!(
            wait_until(|| n.load(Ordering::SeqCst) >= 2, Duration::from_secs(2)),
            "flusher did not tick at least twice"
        );
        f.stop();
        let after = n.load(Ordering::SeqCst);
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(
            n.load(Ordering::SeqCst),
            after,
            "flusher kept ticking after stop()"
        );
    }

    #[test]
    fn flusher_survives_error() {
        let n = Arc::new(AtomicUsize::new(0));
        let n2 = n.clone();
        let mut f = Flusher::start(
            move || {
                let c = n2.fetch_add(1, Ordering::SeqCst);
                if c.is_multiple_of(2) {
                    Err(crate::error::DbError::Config("boom"))
                } else {
                    Ok(())
                }
            },
            Duration::from_millis(10),
        );
        assert!(
            wait_until(|| n.load(Ordering::SeqCst) >= 4, Duration::from_secs(2)),
            "flusher stopped after an error"
        );
        f.stop();
    }
}