armdb 0.14.2

//! Per-tick barrier accounting for the Db flush worker.
//!
//! Counters are thread-local so a pass on this worker is not mixed with
//! compaction/rotate/explicit `flush` on other threads, and so two `Db`s in
//! one process do not zero each other's totals mid-pass.

use std::cell::Cell;
use std::time::{Duration, Instant};

thread_local! {
    static ATTEMPTED: Cell<u64> = const { Cell::new(0) };
    static FAILED: Cell<u64> = const { Cell::new(0) };
    static SKIPPED: Cell<u64> = const { Cell::new(0) };
    static LAST_WARN: Cell<Option<Instant>> = const { Cell::new(None) };
}

const WARN_GAP: Duration = Duration::from_secs(30);

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum WarnKind {
    Slow,
    Failed,
}

fn warning_kind(interval: Duration, elapsed: Duration, failed: u64) -> Option<WarnKind> {
    if elapsed > interval {
        Some(WarnKind::Slow)
    } else if failed > 0 {
        Some(WarnKind::Failed)
    } else {
        None
    }
}

#[cfg_attr(not(feature = "armour"), allow(dead_code))]
pub(crate) fn begin() {
    ATTEMPTED.set(0);
    FAILED.set(0);
    SKIPPED.set(0);
}

pub(crate) fn record_attempt() {
    ATTEMPTED.set(ATTEMPTED.get() + 1);
}

pub(crate) fn record_failure() {
    FAILED.set(FAILED.get() + 1);
}

/// A shard that needed no barrier: already synced (Bitcask) or with nothing
/// unsynced (FixedStore). Counted so a pass reports how much of the shard set
/// it actually had to make durable.
pub(crate) fn record_skip() {
    SKIPPED.set(SKIPPED.get() + 1);
    metrics::counter!("armdb.flush.barriers", "result" => "skipped").increment(1);
}

#[cfg_attr(not(feature = "armour"), allow(dead_code))]
pub(crate) fn finish(interval: Duration, elapsed: Duration) {
    finish_at(interval, elapsed, Instant::now());
}

fn finish_at(interval: Duration, elapsed: Duration, now: Instant) {
    let attempted = ATTEMPTED.get();
    let failed = FAILED.get();
    let skipped = SKIPPED.get();
    // Published on every pass, clean ones included: a gauge only written when
    // something went wrong would keep reporting the last bad pass forever.
    metrics::gauge!("armdb.flush.dirty_shards").set(attempted as f64);
    metrics::gauge!("armdb.flush.unsynced_shards").set(failed as f64);
    let Some(kind) = warning_kind(interval, elapsed, failed) else {
        return;
    };
    if LAST_WARN
        .get()
        .is_some_and(|t| now.saturating_duration_since(t) < WARN_GAP)
    {
        return;
    }
    LAST_WARN.set(Some(now));
    #[cfg(test)]
    EMITTED_WARNINGS.set(EMITTED_WARNINGS.get() + 1);
    if kind == WarnKind::Slow {
        tracing::warn!(
            elapsed_ms = elapsed.as_millis() as u64,
            interval_ms = interval.as_millis() as u64,
            barriers_attempted = attempted,
            barriers_failed = failed,
            barriers_skipped = skipped,
            "flush pass exceeded its interval; power-loss window is interval + 2 × pass duration"
        );
    } else {
        tracing::warn!(
            elapsed_ms = elapsed.as_millis() as u64,
            interval_ms = interval.as_millis() as u64,
            barriers_attempted = attempted,
            barriers_failed = failed,
            barriers_skipped = skipped,
            "flush pass recorded failed barriers; dirty shards will retry next tick"
        );
    }
}

#[cfg(test)]
thread_local! {
    static EMITTED_WARNINGS: Cell<u64> = const { Cell::new(0) };
}

#[cfg(test)]
pub(crate) fn counts_for_test() -> (u64, u64) {
    (ATTEMPTED.get(), FAILED.get())
}

#[cfg(test)]
pub(crate) fn skips_for_test() -> u64 {
    SKIPPED.get()
}

#[cfg(test)]
fn reset_for_test() {
    begin();
    LAST_WARN.set(None);
    EMITTED_WARNINGS.set(0);
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use metrics::{
        Counter, CounterFn, Gauge, GaugeFn, Histogram, Key, KeyName, Metadata, Recorder,
        SharedString, Unit,
    };

    use super::*;

    /// Thread-local recorder: `with_local_recorder` scopes it to this test, so
    /// two flush-pass tests in one process never see each other's emissions.
    #[derive(Default)]
    struct TestRecorder {
        gauges: Arc<Mutex<Vec<(String, f64)>>>,
        counters: Arc<Mutex<Vec<(String, u64)>>>,
    }

    struct GaugeSink(String, Arc<Mutex<Vec<(String, f64)>>>);
    struct CounterSink(String, Arc<Mutex<Vec<(String, u64)>>>);

    impl GaugeFn for GaugeSink {
        fn increment(&self, _value: f64) {}
        fn decrement(&self, _value: f64) {}
        fn set(&self, value: f64) {
            self.1
                .lock()
                .expect("gauge sink")
                .push((self.0.clone(), value));
        }
    }

    impl CounterFn for CounterSink {
        fn increment(&self, value: u64) {
            self.1
                .lock()
                .expect("counter sink")
                .push((self.0.clone(), value));
        }
        fn absolute(&self, _value: u64) {}
    }

    impl Recorder for TestRecorder {
        fn describe_counter(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
        fn describe_gauge(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
        fn describe_histogram(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
        fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> Counter {
            Counter::from_arc(Arc::new(CounterSink(
                key.name().to_owned(),
                Arc::clone(&self.counters),
            )))
        }
        fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> Gauge {
            Gauge::from_arc(Arc::new(GaugeSink(
                key.name().to_owned(),
                Arc::clone(&self.gauges),
            )))
        }
        fn register_histogram(&self, _: &Key, _: &Metadata<'_>) -> Histogram {
            Histogram::noop()
        }
    }

    impl TestRecorder {
        fn gauge(&self, name: &str) -> Option<f64> {
            self.gauges
                .lock()
                .expect("gauge sink")
                .iter()
                .rev()
                .find(|(k, _)| k == name)
                .map(|(_, v)| *v)
        }
    }

    #[test]
    fn finish_publishes_the_loss_window_gauges() {
        reset_for_test();
        record_attempt();
        record_attempt();
        record_failure();

        let recorder = TestRecorder::default();
        metrics::with_local_recorder(&recorder, || {
            finish_at(
                Duration::from_secs(1),
                Duration::from_millis(1),
                Instant::now(),
            );
        });

        assert_eq!(
            recorder.gauge("armdb.flush.dirty_shards"),
            Some(2.0),
            "the pass barriered two dirty shards"
        );
        assert_eq!(
            recorder.gauge("armdb.flush.unsynced_shards"),
            Some(1.0),
            "one failed barrier leaves one shard in the loss window"
        );
    }

    #[test]
    fn finish_publishes_zero_gauges_for_a_clean_pass() {
        reset_for_test();
        record_skip();

        let recorder = TestRecorder::default();
        metrics::with_local_recorder(&recorder, || {
            finish_at(
                Duration::from_secs(1),
                Duration::from_millis(1),
                Instant::now(),
            );
        });

        assert_eq!(
            recorder.gauge("armdb.flush.dirty_shards"),
            Some(0.0),
            "a clean pass must publish zero, not leave a stale gauge"
        );
        assert_eq!(recorder.gauge("armdb.flush.unsynced_shards"), Some(0.0));
    }

    #[test]
    fn skips_are_counted_until_the_next_pass_begins() {
        reset_for_test();
        record_skip();
        record_skip();
        assert_eq!(skips_for_test(), 2);
        begin();
        assert_eq!(
            skips_for_test(),
            0,
            "a new pass starts from zero skipped barriers"
        );
    }

    #[test]
    fn warning_kind_prefers_slow_pass() {
        assert_eq!(
            warning_kind(Duration::from_secs(1), Duration::from_secs(2), 1),
            Some(WarnKind::Slow)
        );
    }

    #[test]
    fn warning_kind_reports_slow_clean_pass() {
        assert_eq!(
            warning_kind(Duration::from_secs(1), Duration::from_secs(2), 0),
            Some(WarnKind::Slow)
        );
    }

    #[test]
    fn warning_kind_reports_failed_pass_within_interval() {
        assert_eq!(
            warning_kind(Duration::from_secs(2), Duration::from_secs(1), 1),
            Some(WarnKind::Failed)
        );
    }

    #[test]
    fn warning_kind_skips_clean_pass() {
        assert_eq!(
            warning_kind(Duration::from_secs(1), Duration::from_secs(1), 0),
            None
        );
    }

    #[test]
    fn flush_pass_suppresses_before_thirty_seconds_and_reemits_at_expiry() {
        reset_for_test();
        let now = Instant::now();
        finish_at(Duration::from_secs(1), Duration::from_secs(2), now);
        finish_at(
            Duration::from_secs(1),
            Duration::from_secs(2),
            now + WARN_GAP - Duration::from_nanos(1),
        );
        assert_eq!(
            EMITTED_WARNINGS.get(),
            1,
            "LAST_WARN suppresses emissions strictly before thirty seconds"
        );
        finish_at(
            Duration::from_secs(1),
            Duration::from_secs(2),
            now + WARN_GAP,
        );
        assert_eq!(
            EMITTED_WARNINGS.get(),
            2,
            "LAST_WARN permits re-emission at the thirty-second boundary"
        );
    }
}