async_nodes/

recovery.rs

//! Per-node failure recovery: retry with backoff, or route to a declarative error edge.
//!
//! The engine owns the *mechanics* of recovery; consumers own the *classification*. The
//! engine knows no HTTP statuses, no provider quirks — a consumer plugs its domain
//! knowledge in through the `classify` hook of [`RecoveryPolicy::Retry`], mapping its
//! errors (wrapped in [`BspError`]) to the generic [`Recovery`] lanes.
//!
//! # What retrying a node means
//!
//! **What the engine retries is the node execution, nothing larger.** Re-invoking
//! `execute` re-runs the node's whole body with the same gathered inputs; whatever the
//! failed attempt wrote into the consumer state or value store is the consumer's rollback
//! concern. A node that wraps side effects (sending email, charging cards) must therefore
//! opt into retry consciously: only read-only nodes, or nodes that implement
//! exactly-once-ish rollback via [`crate::run::RunContext::execution_no`], should carry a
//! retry policy. The type system cannot save you here — `RecoveryPolicy::Retry` on a node
//! wrapping `send_email` sends email twice.

use std::{
    fmt,
    sync::Arc,
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use serde::{Deserialize, Serialize};

use super::error::BspError;

/// How a failed node execution should be recovered, if at all.
///
/// Produced by the consumer's `classify` hook; consumed by the engine's retry loop.
/// Deliberately generic: the engine knows no domain-specific failure lanes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Recovery {
    /// Transient failure — retry after backoff. `after` carries an external hint (e.g. a
    /// server's `Retry-After` header captured by the consumer); the effective sleep is
    /// `max(computed_backoff, after)`.
    Retry { after: Option<Duration> },

    /// Deterministic or unrecoverable failure. Propagate immediately; waiting cannot help.
    Fatal,
}

/// Declarative backoff policy governing automatic retry of failed node executions.
///
/// Domain-free port of the shape battle-tested in the reference consumer (`llm-net`'s
/// `graph/retry.rs`): exponential backoff from `base_delay` to a `max_delay` ceiling with
/// symmetric jitter to prevent lockstep retries across parallel fan-outs.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RetryPolicy {
    /// How many times a failed execution is re-attempted after the initial failure.
    /// `0` disables retry entirely.
    pub max_retries: u32,

    /// The initial backoff delay, doubled per attempt.
    pub base_delay: Duration,

    /// The hard ceiling for any single backoff delay.
    pub max_delay: Duration,

    /// Symmetric randomization factor applied to each delay (e.g. `0.25` = ±25%).
    pub jitter: f64,
}

impl Default for RetryPolicy {
    /// The recommended default: 3 retries, 500ms → 8s exponential backoff, ±25% jitter.
    fn default() -> Self {
        Self {
            max_retries: 3,
            base_delay: Duration::from_millis(500),
            max_delay: Duration::from_secs(8),
            jitter: 0.25,
        }
    }
}

impl RetryPolicy {
    /// A policy that never retries: the first failure propagates.
    #[must_use]
    pub const fn none() -> Self {
        Self {
            max_retries: 0,
            base_delay: Duration::from_millis(500),
            max_delay: Duration::from_secs(8),
            jitter: 0.25,
        }
    }

    /// Overrides the retry budget.
    #[must_use]
    pub const fn with_max_retries(mut self, max_retries: u32) -> Self {
        self.max_retries = max_retries;
        self
    }

    /// Overrides the initial backoff delay.
    #[must_use]
    pub const fn with_base_delay(mut self, base_delay: Duration) -> Self {
        self.base_delay = base_delay;
        self
    }

    /// Overrides the backoff ceiling.
    #[must_use]
    pub const fn with_max_delay(mut self, max_delay: Duration) -> Self {
        self.max_delay = max_delay;
        self
    }

    /// Overrides the jitter factor (`0.0` disables randomization).
    #[must_use]
    pub const fn with_jitter(mut self, jitter: f64) -> Self {
        self.jitter = jitter;
        self
    }

    /// Computes the backoff delay for the given (zero-based) retry attempt:
    /// `min(base * 2^attempt, max) * (1 ± jitter)`.
    #[must_use]
    pub fn backoff(&self, attempt: u32) -> Duration {
        // Clamp the shift: 2^6 already pushes any sane base past its ceiling.
        let exp = self.base_delay.saturating_mul(1 << attempt.min(6));
        let capped = exp.min(self.max_delay);

        if self.jitter <= 0.0 {
            return capped;
        }
        // Cheap time-derived entropy for jitter — keeps the dependency tree clean
        // (no `rand` in the crate) and is more than uniform enough for backoff.
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.subsec_nanos())
            .unwrap_or(0);
        let r = f64::from(nanos % 1000) / 1000.0; // [0, 1)
        let factor = (1.0 + self.jitter * (r * 2.0 - 1.0)).max(0.05);
        capped.mul_f64(factor)
    }
}

/// The consumer hook mapping a domain error (wrapped in [`BspError`]) to a recovery lane.
pub type ClassifyFn = Arc<dyn Fn(&BspError) -> Recovery + Send + Sync>;

/// Per-node recovery policy, declared by the node and enforced by the engine.
///
/// The default is [`RecoveryPolicy::FailFast`]: the first node error aborts the run, which
/// is the behavior dataflow pipelines have always had.
#[derive(Clone)]
pub enum RecoveryPolicy {
    /// First error aborts the run (the default).
    FailFast,

    /// Re-run the node with backoff when the classifier says the failure is transient.
    ///
    /// **Warning:** the engine re-invokes `execute` verbatim. Only attach this to nodes
    /// that are read-only or that undo their partial writes on re-entry (see
    /// [`crate::run::RunContext::execution_no`]); retrying a side-effecting node replays
    /// the side effect.
    Retry {
        /// The backoff policy governing attempts.
        policy: RetryPolicy,
        /// Consumer hook classifying each failure into a [`Recovery`] lane.
        classify: ClassifyFn,
    },

    /// Route the failure to declarative [`crate::condition::Condition::OnError`] edges
    /// instead of aborting: the error is recorded in the run state (and value store) and
    /// the run continues down the error path. GUI-renderable — error handling becomes
    /// visible topology.
    RouteToErrorEdge,
}

impl RecoveryPolicy {
    /// Convenience constructor for [`RecoveryPolicy::Retry`] with a classification hook.
    #[must_use]
    pub fn retry(
        policy: RetryPolicy,
        classify: impl Fn(&BspError) -> Recovery + Send + Sync + 'static,
    ) -> Self {
        Self::Retry {
            policy,
            classify: Arc::new(classify),
        }
    }
}

impl fmt::Debug for RecoveryPolicy {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::FailFast => f.write_str("FailFast"),
            Self::Retry { policy, .. } => f.debug_tuple("Retry").field(policy).finish(),
            Self::RouteToErrorEdge => f.write_str("RouteToErrorEdge"),
        }
    }
}

/// Async sleep used between retry attempts.
///
/// Uses `tokio::time::sleep` when any tokio-enabled crate feature is on. In tokio-free
/// builds there is no runtime timer to hook into, so this falls back to blocking the
/// current thread — correct but unkind to the surrounding executor; enable a tokio feature
/// for production retry use.
pub(crate) async fn sleep(duration: Duration) {
    #[cfg(feature = "tokio")]
    {
        tokio::time::sleep(duration).await;
    }
    #[cfg(not(feature = "tokio"))]
    {
        std::thread::sleep(duration);
    }
}

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

    #[test]
    fn test_default_policy_is_three_retries() {
        let policy = RetryPolicy::default();
        assert_eq!(policy.max_retries, 3);
        assert_eq!(policy.base_delay, Duration::from_millis(500));
        assert_eq!(policy.max_delay, Duration::from_secs(8));
        assert!((policy.jitter - 0.25).abs() < f64::EPSILON);
    }

    #[test]
    fn test_none_disables_retry() {
        assert_eq!(RetryPolicy::none().max_retries, 0);
    }

    #[test]
    fn test_const_builders() {
        let policy = RetryPolicy::default()
            .with_max_retries(5)
            .with_base_delay(Duration::from_millis(10))
            .with_max_delay(Duration::from_secs(1))
            .with_jitter(0.0);
        assert_eq!(policy.max_retries, 5);
        assert_eq!(policy.base_delay, Duration::from_millis(10));
        assert_eq!(policy.max_delay, Duration::from_secs(1));
        assert!(policy.jitter.abs() < f64::EPSILON);
    }

    #[test]
    fn test_backoff_grows_and_respects_bounds() {
        let policy = RetryPolicy::default().with_jitter(0.0);
        assert_eq!(policy.backoff(0), Duration::from_millis(500));
        assert_eq!(policy.backoff(1), Duration::from_secs(1));
        assert_eq!(policy.backoff(2), Duration::from_secs(2));
        // Past the ceiling: capped, never panics on huge attempts.
        assert_eq!(policy.backoff(100), Duration::from_secs(8));
    }

    #[test]
    fn test_backoff_jitter_stays_within_factor() {
        let policy = RetryPolicy::default().with_jitter(0.25);
        for attempt in 0..10 {
            let d = policy.backoff(attempt);
            let base = policy.clone().with_jitter(0.0).backoff(attempt);
            let lo = base.mul_f64(0.74); // 0.75 minus float slack
            let hi = base.mul_f64(1.26);
            assert!(
                d >= lo && d <= hi,
                "attempt {attempt}: {d:?} outside [{lo:?}, {hi:?}]"
            );
        }
    }

    #[test]
    fn test_recovery_policy_retry_constructor() {
        let policy = RecoveryPolicy::retry(RetryPolicy::none(), |_| Recovery::Fatal);
        let RecoveryPolicy::Retry { classify, .. } = policy else {
            panic!("expected Retry variant");
        };
        assert_eq!(classify(&BspError::Deadlock), Recovery::Fatal);
    }
}