hyperdir 0.5.0

//! Opt-in S3 fault injection, for turning intermittent failures into
//! deterministic tests.
//!
//! Several classes of bug here only surface when an S3 request fails in the
//! middle of a multi-request operation — a throttled page of a paginated LIST,
//! for instance, once returned a partial scatter listing that made a deleted
//! name look live. Such bugs cannot be regression-tested by waiting for real
//! throttling, so this interceptor fails chosen operations on demand.
//!
//! It is compiled only with the `fault-injection` feature, so a normal build
//! contains none of this code. Attach it to the S3 client and drive it either
//! programmatically ([`S3FaultInterceptor::new`]) or, for a spawned process such
//! as a mount under test, from the environment via [`S3FaultInterceptor::from_env`]:
//!
//! ```text
//! HYPERDIR_FAULT_OPS=ListObjectsV2   # comma-separated operation names
//! HYPERDIR_FAULT_EVERY=3             # fail every Nth matching request
//! HYPERDIR_FAULT_SKIP=2              # let the first N matching requests through
//! ```

use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

/// Fails selected S3 operations so a caller's error handling can be tested.
///
/// Matching is by operation name (as the SDK reports it, e.g. `ListObjectsV2`,
/// `GetObject`, `PutObject`). `every` failures are deterministic by request
/// count, not random, so a test that injects them is reproducible.
#[derive(Clone)]
pub struct S3FaultInterceptor {
    ops: Arc<Vec<String>>,
    every: u64,
    skip: u64,
    seen: Arc<AtomicU64>,
    injected: Arc<AtomicU64>,
}

impl fmt::Debug for S3FaultInterceptor {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "S3FaultInterceptor(ops={:?}, every={}, skip={})", self.ops, self.every, self.skip)
    }
}

impl S3FaultInterceptor {
    /// Fail every `every`-th request to any of `ops`, after letting the first
    /// `skip` matching requests succeed. `every == 0` disables injection.
    pub fn new(ops: &[&str], every: u64, skip: u64) -> Self {
        Self {
            ops: Arc::new(ops.iter().map(|s| s.to_string()).collect()),
            every,
            skip,
            seen: Arc::new(AtomicU64::new(0)),
            injected: Arc::new(AtomicU64::new(0)),
        }
    }

    /// Build from `HYPERDIR_FAULT_OPS` / `HYPERDIR_FAULT_EVERY` /
    /// `HYPERDIR_FAULT_SKIP`; returns `None` when no ops are configured, so a
    /// caller can attach it only when injection is actually wanted.
    pub fn from_env() -> Option<Self> {
        let ops = std::env::var("HYPERDIR_FAULT_OPS").ok()?;
        let ops: Vec<String> = ops.split(',').map(|s| s.trim().to_string()).filter(|s| !s.is_empty()).collect();
        if ops.is_empty() {
            return None;
        }
        let num = |k: &str, d: u64| std::env::var(k).ok().and_then(|v| v.parse().ok()).unwrap_or(d);
        Some(Self {
            ops: Arc::new(ops),
            every: num("HYPERDIR_FAULT_EVERY", 1),
            skip: num("HYPERDIR_FAULT_SKIP", 0),
            seen: Arc::new(AtomicU64::new(0)),
            injected: Arc::new(AtomicU64::new(0)),
        })
    }

    /// How many faults have been injected so far (for test assertions).
    pub fn injected(&self) -> u64 {
        self.injected.load(Ordering::Relaxed)
    }

    /// How many matching requests have been seen so far.
    pub fn seen(&self) -> u64 {
        self.seen.load(Ordering::Relaxed)
    }
}

impl aws_sdk_s3::config::Intercept for S3FaultInterceptor {
    fn name(&self) -> &'static str {
        "S3FaultInterceptor"
    }

    /// Fail just before the request goes out: the operation sees a transport-level
    /// error, which is what a reset connection or a throttled retry exhaustion
    /// looks like to the caller.
    fn modify_before_transmit(
        &self,
        _ctx: &mut aws_sdk_s3::config::interceptors::BeforeTransmitInterceptorContextMut<'_>,
        _: &aws_sdk_s3::config::RuntimeComponents,
        cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
        if self.every == 0 {
            return Ok(());
        }
        let op = cfg
            .load::<aws_smithy_runtime_api::client::orchestrator::Metadata>()
            .map(|m| m.name().to_string())
            .unwrap_or_default();
        if !self.ops.iter().any(|o| o == &op) {
            return Ok(());
        }
        let n = self.seen.fetch_add(1, Ordering::Relaxed) + 1;
        if n <= self.skip || !(n - self.skip).is_multiple_of(self.every) {
            return Ok(());
        }
        self.injected.fetch_add(1, Ordering::Relaxed);
        Err(format!("injected fault: {} request #{}", op, n).into())
    }
}