batpak 0.10.0

//! Fault injection framework for testing failure scenarios.
//!
//! This module provides structured fault injection for testing recovery,
//! error handling paths, and partial-write scenarios that are otherwise
//! hard to trigger in normal testing.
//!
//! # Usage
//!
//! ```rust,ignore
//! use batpak::store::fault::{FaultInjector, InjectionPoint};
//! use batpak::store::StoreError;
//!
//! struct FailAfterBegin;
//! impl FaultInjector for FailAfterBegin {
//!     fn check(&self, point: InjectionPoint) -> Option<StoreError> {
//!         if matches!(point, InjectionPoint::BatchBeginWritten { .. }) {
//!             Some(StoreError::FaultInjected(
//!                 format!("simulated failure at {point:?}"),
//!             ))
//!         } else {
//!             None
//!         }
//!     }
//! }
//!
//! let config = StoreConfig::new(path).with_fault_injector(Some(Arc::new(FailAfterBegin)));
//! ```

use crate::store::StoreError;
use std::sync::Arc;

type InjectionPointFilter = Box<dyn Fn(&InjectionPoint) -> bool + Send + Sync>;

/// Injection points in the writer where faults can be triggered.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum InjectionPoint {
    /// Before writing any batch frames.
    BatchStart {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
        /// Number of items in this batch.
        item_count: usize,
    },

    /// After writing the BEGIN marker but before any items.
    BatchBeginWritten {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
        /// Number of items in this batch.
        item_count: usize,
    },

    /// After writing N items in a batch.
    BatchItemWritten {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
        /// Zero-based index of the item just written.
        item_index: usize,
        /// Total number of items in the batch.
        total_items: usize,
    },

    /// After writing all items but before the COMMIT marker.
    BatchItemsComplete {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
        /// Number of items in this batch.
        item_count: usize,
    },

    /// After writing the COMMIT marker but before fsync.
    BatchCommitWritten {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
    },

    /// During fsync after COMMIT marker.
    BatchFsync {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
    },

    /// After successful fsync, before index publish.
    BatchPrePublish {
        /// Monotonic batch identifier from global sequence.
        batch_id: u64,
        /// Number of items in this batch.
        item_count: usize,
    },

    /// Single event append before write.
    SingleAppendStart {
        /// Entity name for the event being appended.
        entity: String,
    },

    /// Single event append after write, before fsync.
    SingleAppendWritten {
        /// Entity name for the event just written.
        entity: String,
    },

    /// Single event append after publish/broadcast, before returning receipt.
    SingleAppendPublished {
        /// Entity name for the event just made visible.
        entity: String,
    },

    /// During segment rotation, BEFORE the new active segment is created.
    ///
    /// Models a failure of `Segment::create_with_created_ns` (the new-segment
    /// file create + file/dir fsync) at the very start of rotation, while the
    /// old segment and SIDX collector are still pristine. A fault here must
    /// abort the rotation with the old segment + collector fully intact (no
    /// SIDX footer written, collector unchanged), so the writer state stays
    /// consistent and the next append cleanly retries against the unchanged
    /// old segment.
    SegmentRotationCreate {
        /// Segment being sealed.
        old_segment: u64,
        /// Segment being opened.
        new_segment: u64,
    },

    /// During segment rotation, AFTER in-memory state is fully rolled forward
    /// (old segment sealed with footer, new active file present, segment_id
    /// advanced, reader notified).
    SegmentRotation {
        /// Segment being sealed.
        old_segment: u64,
        /// Segment being opened.
        new_segment: u64,
    },

    /// During the writer's active-segment durability fsync — the single choke
    /// point every durable-frontier advance routes through (periodic cadence
    /// sync, explicit `Sync` barrier, segment rotation's data flush,
    /// batch-commit fsync, and the shutdown drain's final sync). Fired BEFORE
    /// the real fsync; an injected error is handled exactly like a real fsync
    /// failure: it permanently poisons the writer fail-closed (fsyncgate — a
    /// later Ok fsync cannot vouch for pages the kernel dropped after an
    /// fsync error), so every subsequent command surfaces
    /// `StoreError::WriterCrashed` and the durable frontier never advances
    /// past the last successful sync.
    ActiveSegmentSync {
        /// Active segment being fsynced.
        segment_id: u64,
    },

    // ---------------------------------------------------------------------
    // READ / SCAN / COLD-START injection points (GAUNT-FAULT-3).
    //
    // These mirror the write-path call-site pattern but cover the recovery
    // surface exercised during `Store::open` / `Store::open_read_only`: mmap
    // fast-path load, checkpoint decode, sealed-segment SIDX footer decode,
    // per-frame segment scan, and hidden-ranges (cancelled visibility) load.
    // A fault at any of these must leave the store either consistently opened
    // or aborted with a typed `StoreError` (never a panic / partial state).
    // ---------------------------------------------------------------------
    /// Before attempting to load the mmap index fast-path snapshot
    /// (`index.fbati`) during cold start.
    MmapIndexLoad,

    /// While decoding the mmap index footer/header (after the file is mapped,
    /// before its contents are trusted).
    IndexFooterDecode {
        /// Segment whose SIDX footer is being decoded, if known.
        segment_id: u64,
    },

    /// Before attempting to decode the checkpoint snapshot (`index.ckpt`)
    /// during cold start.
    CheckpointDecode,

    /// Before loading the persisted hidden (cancelled visibility) ranges
    /// during cold start. A fault here must fail closed rather than silently
    /// resurrect cancelled events.
    HiddenRangesLoad,

    /// Per-frame during a cold-start segment scan (frame-scan fallback or
    /// tail recovery). Triggered once per scanned frame.
    ColdStartScanFrame {
        /// Segment being scanned.
        segment_id: u64,
        /// Zero-based index of the frame within this scan pass.
        frame_index: usize,
    },

    /// A low-level positional read during recovery. Generic catch-all for
    /// `read_at`-style I/O on the cold-start path.
    ReadAt {
        /// Byte offset of the read.
        offset: u64,
        /// Number of bytes requested.
        len: usize,
    },
}

/// Trait for implementing fault injection scenarios.
///
/// Implementors inspect each [`InjectionPoint`] and return
/// `Some(StoreError)` to inject a fault, or `None` to proceed normally.
pub trait FaultInjector: Send + Sync {
    /// Inspect an injection point and optionally return an error to inject.
    ///
    /// Return `None` to let the operation proceed. Return
    /// `Some(StoreError::FaultInjected(..))` (or any `StoreError` variant)
    /// to abort the current operation with that error.
    fn check(&self, point: InjectionPoint) -> Option<StoreError>;

    /// Optional: called when injector is registered to verify configuration.
    ///
    /// # Errors
    /// Returns `Err(String)` if the injector configuration is invalid.
    fn validate(&self) -> Result<(), String> {
        Ok(())
    }
}

/// A fault injector that triggers at a specific sequence of points.
///
/// Useful for testing "crash at Nth operation" scenarios.
pub struct CountdownInjector {
    /// Total number of matching points to wait for before triggering.
    trigger_after: usize,
    /// Current count of matching points seen.
    current: std::sync::atomic::AtomicUsize,
    /// The point type to count, or None for any point.
    filter: Option<InjectionPointFilter>,
    /// Action to take when triggering.
    action: CountdownAction,
}

/// Action to take when a [`CountdownInjector`] or [`ProbabilisticInjector`] triggers.
#[derive(Clone, Copy, Debug)]
pub enum CountdownAction {
    /// Return a [`StoreError::FaultInjected`] with the given message.
    Fail(&'static str),
    /// Count the point but do not inject a fault.
    Noop,
}

impl CountdownInjector {
    /// Create a new countdown injector that triggers after N occurrences.
    pub fn new(trigger_after: usize, action: CountdownAction) -> Self {
        Self {
            trigger_after,
            current: std::sync::atomic::AtomicUsize::new(0),
            filter: None,
            action,
        }
    }

    /// Add a filter so only specific points are counted.
    pub fn with_filter<F>(mut self, filter: F) -> Self
    where
        F: Fn(&InjectionPoint) -> bool + Send + Sync + 'static,
    {
        self.filter = Some(Box::new(filter));
        self
    }

    /// Convenience: fail after N batch items written.
    pub fn after_batch_items(n: usize) -> Self {
        Self::new(
            n,
            CountdownAction::Fail("simulated fault during batch item write"),
        )
        .with_filter(|p| matches!(p, InjectionPoint::BatchItemWritten { .. }))
    }

    /// Convenience: fail after BEGIN marker written.
    pub fn after_batch_begin() -> Self {
        Self::new(
            1,
            CountdownAction::Fail("simulated fault after BEGIN marker"),
        )
        .with_filter(|p| matches!(p, InjectionPoint::BatchBeginWritten { .. }))
    }

    /// Convenience: fail after COMMIT marker but before fsync.
    pub fn after_commit_before_fsync() -> Self {
        Self::new(
            1,
            CountdownAction::Fail("simulated fault after COMMIT before fsync"),
        )
        .with_filter(|p| matches!(p, InjectionPoint::BatchCommitWritten { .. }))
    }

    /// Convenience: fail on the Kth read/scan/cold-start I/O point.
    ///
    /// Counts only recovery-path injection points (mmap load, checkpoint
    /// decode, SIDX footer decode, per-frame scan, hidden-ranges load, and
    /// positional reads), so the same injector can be reused across the whole
    /// cold-start surface to model "crash on the Kth recovery I/O".
    pub fn on_kth_recovery_io(k: usize) -> Self {
        Self::new(
            k,
            CountdownAction::Fail("simulated fault on Kth recovery I/O"),
        )
        .with_filter(|p| {
            matches!(
                p,
                InjectionPoint::MmapIndexLoad
                    | InjectionPoint::IndexFooterDecode { .. }
                    | InjectionPoint::CheckpointDecode
                    | InjectionPoint::HiddenRangesLoad
                    | InjectionPoint::ColdStartScanFrame { .. }
                    | InjectionPoint::ReadAt { .. }
            )
        })
    }
}

impl FaultInjector for CountdownInjector {
    fn check(&self, point: InjectionPoint) -> Option<StoreError> {
        let dominated = self.filter.as_ref().is_none_or(|f| f(&point));
        if !dominated {
            return None;
        }

        let count = self
            .current
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        if count + 1 < self.trigger_after {
            return None;
        }

        match self.action {
            CountdownAction::Fail(msg) => {
                Some(StoreError::FaultInjected(format!("{msg} at {point:?}")))
            }
            CountdownAction::Noop => {
                tracing::debug!("FaultInjector noop at {point:?}");
                None
            }
        }
    }
}

/// A fault injector that triggers based on a probability.
///
/// Useful for chaos testing scenarios.
pub struct ProbabilisticInjector {
    probability: f64,
    filter: Option<InjectionPointFilter>,
    action: CountdownAction,
}

impl ProbabilisticInjector {
    /// Create a new probabilistic injector.
    /// probability: 0.0 to 1.0
    pub fn new(probability: f64, action: CountdownAction) -> Self {
        assert!(
            (0.0..=1.0).contains(&probability),
            "probability must be in [0.0, 1.0]"
        );
        Self {
            probability,
            filter: None,
            action,
        }
    }

    /// Add a filter so only specific points can trigger.
    pub fn with_filter<F>(mut self, filter: F) -> Self
    where
        F: Fn(&InjectionPoint) -> bool + Send + Sync + 'static,
    {
        self.filter = Some(Box::new(filter));
        self
    }
}

impl FaultInjector for ProbabilisticInjector {
    fn check(&self, point: InjectionPoint) -> Option<StoreError> {
        let dominated = self.filter.as_ref().is_none_or(|f| f(&point));
        if !dominated {
            return None;
        }

        let mut rng = fastrand::Rng::new();
        if rng.f64() >= self.probability {
            return None;
        }

        match self.action {
            CountdownAction::Fail(msg) => {
                Some(StoreError::FaultInjected(format!("{msg} at {point:?}")))
            }
            CountdownAction::Noop => {
                tracing::debug!("ProbabilisticInjector noop at {point:?}");
                None
            }
        }
    }
}

/// Check the injection point and propagate any injected fault as an error.
///
/// Returns `Ok(())` if no fault is injected, or `Err(StoreError)` if the
/// injector decided to inject a fault at this point.
///
/// # Errors
/// Returns the `StoreError` produced by the injector's [`FaultInjector::check`].
pub fn maybe_inject(
    point: InjectionPoint,
    injector: &Option<Arc<dyn FaultInjector>>,
) -> Result<(), StoreError> {
    if let Some(inj) = injector {
        if let Some(err) = inj.check(point) {
            return Err(err);
        }
    }
    Ok(())
}

#[cfg(test)]
#[path = "fault_tests.rs"]
mod contract_tests;

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

    #[test]
    fn countdown_triggers_at_count() {
        let injector = CountdownInjector::new(3, CountdownAction::Fail("boom"));

        let point = InjectionPoint::BatchItemWritten {
            batch_id: 1,
            item_index: 0,
            total_items: 5,
        };

        assert!(injector.check(point.clone()).is_none());
        assert!(injector.check(point.clone()).is_none());
        assert!(injector.check(point).is_some()); // 3rd time triggers
    }

    #[test]
    fn countdown_noop_never_faults() {
        let injector = CountdownInjector::new(1, CountdownAction::Noop);

        let point = InjectionPoint::BatchItemWritten {
            batch_id: 1,
            item_index: 0,
            total_items: 5,
        };

        // Noop counts but never returns an error.
        assert!(injector.check(point.clone()).is_none());
        assert!(injector.check(point).is_none());
    }

    #[test]
    fn countdown_respects_filter() {
        let injector = CountdownInjector::new(1, CountdownAction::Fail("boom"))
            .with_filter(|p| matches!(p, InjectionPoint::BatchBeginWritten { .. }));

        let item_point = InjectionPoint::BatchItemWritten {
            batch_id: 1,
            item_index: 0,
            total_items: 5,
        };
        let begin_point = InjectionPoint::BatchBeginWritten {
            batch_id: 1,
            item_count: 5,
        };

        assert!(injector.check(item_point).is_none()); // filtered out
        assert!(injector.check(begin_point).is_some()); // matches filter
    }

    #[test]
    fn fault_injected_error_is_store_error() {
        let injector = CountdownInjector::new(1, CountdownAction::Fail("test fault"));
        let point = InjectionPoint::BatchStart {
            batch_id: 42,
            item_count: 3,
        };
        let err = injector.check(point).expect("should produce error");
        assert!(
            matches!(err, StoreError::FaultInjected(_)),
            "expected FaultInjected variant"
        );
    }

    #[test]
    fn probabilistic_injector_is_deterministic_at_probability_extremes() {
        let point = InjectionPoint::BatchCommitWritten { batch_id: 7 };

        let never = ProbabilisticInjector::new(0.0, CountdownAction::Fail("boom"));
        assert!(
            never.check(point.clone()).is_none(),
            "probability=0.0 must never inject a fault"
        );

        let always = ProbabilisticInjector::new(1.0, CountdownAction::Fail("boom"));
        assert!(
            matches!(always.check(point), Some(StoreError::FaultInjected(_))),
            "probability=1.0 must always inject the configured fault"
        );
    }

    #[test]
    fn maybe_inject_propagates_injector_decision() {
        let injector: Arc<dyn FaultInjector> =
            Arc::new(CountdownInjector::new(1, CountdownAction::Fail("boom")));
        let point = InjectionPoint::BatchStart {
            batch_id: 9,
            item_count: 2,
        };

        let err = maybe_inject(point, &Some(injector)).expect_err("fault should propagate");
        assert!(
            matches!(err, StoreError::FaultInjected(_)),
            "maybe_inject must return the injector-produced StoreError variant"
        );
    }
}