From fc5821c880262b603bb7b91a46ea06d63e485fd5 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 24 Mar 2026 05:12:48 -0400 Subject: [PATCH] Consolidate backoff/jitter into fabro-util::BackoffPolicy Three crates independently implemented the same exponential-backoff-with-jitter logic. Extract a single BackoffPolicy into fabro-util and have fabro-core, fabro-workflows, and fabro-llm all use it, eliminating duplication and making the backoff conversion in core_adapter trivial. Co-Authored-By: Claude Opus 4.6 (1M context) --- Cargo.lock | 3 +- lib/crates/fabro-agent/src/session.rs | 2 +- lib/crates/fabro-core/Cargo.toml | 2 +- lib/crates/fabro-core/src/retry.rs | 104 +-------- lib/crates/fabro-llm/src/generate.rs | 14 +- lib/crates/fabro-llm/src/retry.rs | 59 +++-- lib/crates/fabro-llm/src/types.rs | 93 ++++---- lib/crates/fabro-util/Cargo.toml | 1 + lib/crates/fabro-util/src/backoff.rs | 121 +++++++++++ lib/crates/fabro-util/src/lib.rs | 1 + .../src/core_adapter/handler.rs | 10 +- .../src/core_adapter/lifecycle.rs | 8 - lib/crates/fabro-workflows/src/engine.rs | 202 ++++-------------- 13 files changed, 264 insertions(+), 356 deletions(-) create mode 100644 lib/crates/fabro-util/src/backoff.rs diff --git a/Cargo.lock b/Cargo.lock index 053af41ec..0111c74fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1408,7 +1408,7 @@ name = "fabro-core" version = "0.176.2" dependencies = [ "async-trait", - "rand 0.8.5", + "fabro-util", "serde", "serde_json", "thiserror 2.0.18", @@ -1713,6 +1713,7 @@ dependencies = [ "console 0.15.11", "dirs", "insta", + "rand 0.8.5", "regex", "serde", "serde_json", diff --git a/lib/crates/fabro-agent/src/session.rs b/lib/crates/fabro-agent/src/session.rs index 64fed19c6..bca746b9c 100644 --- a/lib/crates/fabro-agent/src/session.rs +++ b/lib/crates/fabro-agent/src/session.rs @@ -670,7 +670,7 @@ impl Session { provider: retry_provider.clone(), model: retry_model.clone(), attempt: attempt as usize, - delay_secs: delay, + delay_secs: delay.as_secs_f64(), error: err.clone(), }, ); diff --git a/lib/crates/fabro-core/Cargo.toml b/lib/crates/fabro-core/Cargo.toml index ce4bfd9a2..c5bb84899 100644 --- a/lib/crates/fabro-core/Cargo.toml +++ b/lib/crates/fabro-core/Cargo.toml @@ -10,10 +10,10 @@ doctest = false [dependencies] async-trait.workspace = true +fabro-util = { path = "../fabro-util" } serde.workspace = true serde_json.workspace = true thiserror.workspace = true -rand.workspace = true tokio.workspace = true tokio-util.workspace = true tracing.workspace = true diff --git a/lib/crates/fabro-core/src/retry.rs b/lib/crates/fabro-core/src/retry.rs index 07a603a62..a13cf0082 100644 --- a/lib/crates/fabro-core/src/retry.rs +++ b/lib/crates/fabro-core/src/retry.rs @@ -1,44 +1,4 @@ -use std::time::Duration; - -use rand::Rng; - -#[derive(Debug, Clone)] -pub struct BackoffPolicy { - pub initial_delay: Duration, - pub factor: f64, - pub max_delay: Duration, - pub jitter: bool, -} - -impl Default for BackoffPolicy { - fn default() -> Self { - Self { - initial_delay: Duration::from_secs(1), - factor: 2.0, - max_delay: Duration::from_secs(60), - jitter: false, - } - } -} - -impl BackoffPolicy { - pub fn delay_for_attempt(&self, attempt: u32) -> Duration { - let multiplier = self.factor.powi(attempt.saturating_sub(1) as i32); - let base_delay = self.initial_delay.mul_f64(multiplier); - let capped = if base_delay > self.max_delay { - self.max_delay - } else { - base_delay - }; - if self.jitter { - // Apply jitter: random factor in [0.5, 1.5) - let jitter_factor = rand::thread_rng().gen_range(0.5..1.5); - capped.mul_f64(jitter_factor) - } else { - capped - } - } -} +pub use fabro_util::backoff::BackoffPolicy; #[derive(Debug, Clone)] pub struct RetryPolicy { @@ -72,71 +32,9 @@ impl Default for RetryPolicy { mod tests { use super::*; - #[test] - fn backoff_delay_first_attempt() { - let b = BackoffPolicy { - initial_delay: Duration::from_millis(100), - factor: 2.0, - max_delay: Duration::from_secs(10), - jitter: false, - }; - assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); - } - - #[test] - fn backoff_delay_exponential() { - let b = BackoffPolicy { - initial_delay: Duration::from_millis(100), - factor: 2.0, - max_delay: Duration::from_secs(10), - jitter: false, - }; - assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); - assert_eq!(b.delay_for_attempt(3), Duration::from_millis(400)); - assert_eq!(b.delay_for_attempt(4), Duration::from_millis(800)); - } - - #[test] - fn backoff_delay_capped_at_max() { - let b = BackoffPolicy { - initial_delay: Duration::from_millis(100), - factor: 2.0, - max_delay: Duration::from_millis(300), - jitter: false, - }; - assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); - assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); - assert_eq!(b.delay_for_attempt(3), Duration::from_millis(300)); // capped - assert_eq!(b.delay_for_attempt(4), Duration::from_millis(300)); // still capped - } - #[test] fn retry_policy_none_is_single_attempt() { let p = RetryPolicy::none(); assert_eq!(p.max_attempts, 1); } - - #[test] - fn backoff_delay_with_jitter_within_range() { - let b = BackoffPolicy { - initial_delay: Duration::from_millis(1000), - factor: 1.0, - max_delay: Duration::from_secs(10), - jitter: true, - }; - let base = Duration::from_millis(1000); - let min = base.mul_f64(0.5); - let max = base.mul_f64(1.5); - - for _ in 0..100 { - let delay = b.delay_for_attempt(1); - assert!( - delay >= min && delay <= max, - "delay {:?} out of range [{:?}, {:?}]", - delay, - min, - max, - ); - } - } } diff --git a/lib/crates/fabro-llm/src/generate.rs b/lib/crates/fabro-llm/src/generate.rs index f868a7d4c..44cf6eb1f 100644 --- a/lib/crates/fabro-llm/src/generate.rs +++ b/lib/crates/fabro-llm/src/generate.rs @@ -109,8 +109,11 @@ pub async fn generate(params: GenerateParams) -> Result Result policy.max_delay { + let retry_after_dur = Duration::from_secs_f64(retry_after); + if retry_after_dur > policy.backoff.max_delay { return Err(err); } - retry_after + retry_after_dur } else { - policy.delay_for_attempt(attempt) + // Convert from 0-indexed (fabro-llm convention) to 1-indexed (BackoffPolicy) + policy.backoff.delay_for_attempt(attempt + 1) }; warn!( attempt = attempt, - delay_secs = delay, + delay_secs = delay.as_secs_f64(), error = %err, "LLM request failed, retrying" ); @@ -48,7 +51,7 @@ where on_retry(&err, attempt, delay); } - tokio::time::sleep(std::time::Duration::from_secs_f64(delay)).await; + tokio::time::sleep(delay).await; attempt += 1; } @@ -60,14 +63,27 @@ where mod tests { use super::*; use crate::types::RetryPolicy; + use fabro_util::backoff::BackoffPolicy; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; + fn fast_backoff() -> BackoffPolicy { + BackoffPolicy { + initial_delay: Duration::from_micros(1), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + } + } + #[tokio::test] async fn retry_succeeds_first_try() { let policy = RetryPolicy { max_retries: 2, - jitter: false, + backoff: BackoffPolicy { + jitter: false, + ..BackoffPolicy::default() + }, ..Default::default() }; @@ -91,8 +107,7 @@ mod tests { async fn retry_succeeds_after_retries() { let policy = RetryPolicy { max_retries: 3, - base_delay: 0.001, - jitter: false, + backoff: fast_backoff(), ..Default::default() }; @@ -126,8 +141,7 @@ mod tests { async fn retry_gives_up_after_max_retries() { let policy = RetryPolicy { max_retries: 2, - base_delay: 0.001, - jitter: false, + backoff: fast_backoff(), ..Default::default() }; @@ -157,8 +171,7 @@ mod tests { async fn retry_does_not_retry_non_retryable() { let policy = RetryPolicy { max_retries: 3, - base_delay: 0.001, - jitter: false, + backoff: fast_backoff(), ..Default::default() }; @@ -188,9 +201,12 @@ mod tests { async fn retry_skips_when_retry_after_exceeds_max_delay() { let policy = RetryPolicy { max_retries: 3, - base_delay: 0.001, - max_delay: 5.0, - jitter: false, + backoff: BackoffPolicy { + initial_delay: Duration::from_micros(1), + factor: 2.0, + max_delay: Duration::from_secs(5), + jitter: false, + }, ..Default::default() }; @@ -221,9 +237,12 @@ mod tests { async fn retry_uses_retry_after_when_within_limit() { let policy = RetryPolicy { max_retries: 1, - base_delay: 10.0, // base_delay is high, but retry_after is low - max_delay: 60.0, - jitter: false, + backoff: BackoffPolicy { + initial_delay: Duration::from_secs(10), // high, but retry_after is low + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + }, ..Default::default() }; @@ -265,12 +284,10 @@ mod tests { let policy = RetryPolicy { max_retries: 2, - base_delay: 0.001, - jitter: false, + backoff: fast_backoff(), on_retry: Some(Arc::new(move |_err, _attempt, _delay| { retry_attempts_clone.fetch_add(1, Ordering::SeqCst); })), - ..Default::default() }; let call_count = Arc::new(AtomicU32::new(0)); diff --git a/lib/crates/fabro-llm/src/types.rs b/lib/crates/fabro-llm/src/types.rs index 9c161c111..005b1860c 100644 --- a/lib/crates/fabro-llm/src/types.rs +++ b/lib/crates/fabro-llm/src/types.rs @@ -727,17 +727,14 @@ impl Default for AdapterTimeout { // --- 6.6 RetryPolicy --- -/// Callback invoked before each retry attempt with (error, attempt, delay in seconds). -pub type OnRetryCallback = Arc; +/// Callback invoked before each retry attempt with (error, attempt, delay as Duration). +pub type OnRetryCallback = Arc; #[derive(Clone)] pub struct RetryPolicy { pub max_retries: u32, - pub base_delay: f64, - pub max_delay: f64, - pub backoff_multiplier: f64, - pub jitter: bool, - /// Called before each retry with (error, attempt number, delay in seconds). + pub backoff: fabro_util::backoff::BackoffPolicy, + /// Called before each retry with (error, attempt number, delay). pub on_retry: Option, } @@ -745,10 +742,7 @@ impl std::fmt::Debug for RetryPolicy { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("RetryPolicy") .field("max_retries", &self.max_retries) - .field("base_delay", &self.base_delay) - .field("max_delay", &self.max_delay) - .field("backoff_multiplier", &self.backoff_multiplier) - .field("jitter", &self.jitter) + .field("backoff", &self.backoff) .field("on_retry", &self.on_retry.as_ref().map(|_| "...")) .finish() } @@ -758,30 +752,17 @@ impl Default for RetryPolicy { fn default() -> Self { Self { max_retries: 2, - base_delay: 1.0, - max_delay: 60.0, - backoff_multiplier: 2.0, - jitter: true, + backoff: fabro_util::backoff::BackoffPolicy { + initial_delay: std::time::Duration::from_secs(1), + factor: 2.0, + max_delay: std::time::Duration::from_secs(60), + jitter: true, + }, on_retry: None, } } } -impl RetryPolicy { - #[must_use] - pub fn delay_for_attempt(&self, attempt: u32) -> f64 { - let delay = self.base_delay * self.backoff_multiplier.powi(attempt as i32); - let delay = delay.min(self.max_delay); - - if self.jitter { - let jitter_factor = 0.5 + rand::random::(); // 0.5..1.5 - delay * jitter_factor - } else { - delay - } - } -} - // --- 4.6 ObjectStreamEvent --- /// Events yielded by `stream_object()` for streaming structured output. @@ -1177,47 +1158,57 @@ mod tests { #[test] fn retry_policy_delay_no_jitter() { + use std::time::Duration; let policy = RetryPolicy { max_retries: 3, - base_delay: 1.0, - max_delay: 60.0, - backoff_multiplier: 2.0, - jitter: false, + backoff: fabro_util::backoff::BackoffPolicy { + initial_delay: Duration::from_secs(1), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + }, ..Default::default() }; - assert!((policy.delay_for_attempt(0) - 1.0).abs() < f64::EPSILON); - assert!((policy.delay_for_attempt(1) - 2.0).abs() < f64::EPSILON); - assert!((policy.delay_for_attempt(2) - 4.0).abs() < f64::EPSILON); - assert!((policy.delay_for_attempt(3) - 8.0).abs() < f64::EPSILON); + // BackoffPolicy is 1-indexed: attempt 1 = base, attempt 2 = base*factor, etc. + assert_eq!(policy.backoff.delay_for_attempt(1), Duration::from_secs(1)); + assert_eq!(policy.backoff.delay_for_attempt(2), Duration::from_secs(2)); + assert_eq!(policy.backoff.delay_for_attempt(3), Duration::from_secs(4)); + assert_eq!(policy.backoff.delay_for_attempt(4), Duration::from_secs(8)); } #[test] fn retry_policy_delay_respects_max() { + use std::time::Duration; let policy = RetryPolicy { max_retries: 10, - base_delay: 1.0, - max_delay: 5.0, - backoff_multiplier: 2.0, - jitter: false, + backoff: fabro_util::backoff::BackoffPolicy { + initial_delay: Duration::from_secs(1), + factor: 2.0, + max_delay: Duration::from_secs(5), + jitter: false, + }, ..Default::default() }; - assert!((policy.delay_for_attempt(5) - 5.0).abs() < f64::EPSILON); + assert_eq!(policy.backoff.delay_for_attempt(6), Duration::from_secs(5)); } #[test] fn retry_policy_delay_with_jitter_in_range() { + use std::time::Duration; let policy = RetryPolicy { max_retries: 3, - base_delay: 1.0, - max_delay: 60.0, - backoff_multiplier: 2.0, - jitter: true, + backoff: fabro_util::backoff::BackoffPolicy { + initial_delay: Duration::from_secs(1), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: true, + }, ..Default::default() }; - let delay = policy.delay_for_attempt(0); - // base * 0.5 to base * 1.5 => 0.5 to 1.5 - assert!(delay >= 0.5); - assert!(delay <= 1.5); + let delay = policy.backoff.delay_for_attempt(1); + // base * 0.5 to base * 1.5 => 0.5s to 1.5s + assert!(delay >= Duration::from_millis(500)); + assert!(delay <= Duration::from_millis(1500)); } #[test] diff --git a/lib/crates/fabro-util/Cargo.toml b/lib/crates/fabro-util/Cargo.toml index e7f416ba9..b67fe905e 100644 --- a/lib/crates/fabro-util/Cargo.toml +++ b/lib/crates/fabro-util/Cargo.toml @@ -10,6 +10,7 @@ doctest = false [dependencies] console.workspace = true +rand.workspace = true regex.workspace = true termimad.workspace = true aho-corasick.workspace = true diff --git a/lib/crates/fabro-util/src/backoff.rs b/lib/crates/fabro-util/src/backoff.rs new file mode 100644 index 000000000..8df4c9727 --- /dev/null +++ b/lib/crates/fabro-util/src/backoff.rs @@ -0,0 +1,121 @@ +use std::time::Duration; + +use rand::Rng; + +#[derive(Debug, Clone)] +pub struct BackoffPolicy { + pub initial_delay: Duration, + pub factor: f64, + pub max_delay: Duration, + pub jitter: bool, +} + +impl Default for BackoffPolicy { + fn default() -> Self { + Self { + initial_delay: Duration::from_secs(1), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + } + } +} + +impl BackoffPolicy { + pub fn delay_for_attempt(&self, attempt: u32) -> Duration { + let multiplier = self.factor.powi(attempt.saturating_sub(1) as i32); + let base_delay = self.initial_delay.mul_f64(multiplier); + let capped = if base_delay > self.max_delay { + self.max_delay + } else { + base_delay + }; + if self.jitter { + // Apply jitter: random factor in [0.5, 1.5) + let jitter_factor = rand::thread_rng().gen_range(0.5..1.5); + capped.mul_f64(jitter_factor) + } else { + capped + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn delay_first_attempt() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_secs(10), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); + } + + #[test] + fn delay_exponential() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_secs(10), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); + assert_eq!(b.delay_for_attempt(3), Duration::from_millis(400)); + assert_eq!(b.delay_for_attempt(4), Duration::from_millis(800)); + } + + #[test] + fn delay_capped_at_max() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_millis(300), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); + assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); + assert_eq!(b.delay_for_attempt(3), Duration::from_millis(300)); // capped + assert_eq!(b.delay_for_attempt(4), Duration::from_millis(300)); // still capped + } + + #[test] + fn delay_with_jitter_within_range() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(1000), + factor: 1.0, + max_delay: Duration::from_secs(10), + jitter: true, + }; + let base = Duration::from_millis(1000); + let min = base.mul_f64(0.5); + let max = base.mul_f64(1.5); + + for _ in 0..100 { + let delay = b.delay_for_attempt(1); + assert!( + delay >= min && delay <= max, + "delay {:?} out of range [{:?}, {:?}]", + delay, + min, + max, + ); + } + } + + #[test] + fn delay_linear_factor() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(500), + factor: 1.0, + max_delay: Duration::from_secs(60), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(1), Duration::from_millis(500)); + assert_eq!(b.delay_for_attempt(2), Duration::from_millis(500)); + assert_eq!(b.delay_for_attempt(3), Duration::from_millis(500)); + } +} diff --git a/lib/crates/fabro-util/src/lib.rs b/lib/crates/fabro-util/src/lib.rs index be19d465a..f55f9500a 100644 --- a/lib/crates/fabro-util/src/lib.rs +++ b/lib/crates/fabro-util/src/lib.rs @@ -1,3 +1,4 @@ +pub mod backoff; pub mod check_report; pub mod env; pub mod path; diff --git a/lib/crates/fabro-workflows/src/core_adapter/handler.rs b/lib/crates/fabro-workflows/src/core_adapter/handler.rs index b6bd82b3d..1f3960653 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/handler.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/handler.rs @@ -1,7 +1,6 @@ use std::panic::AssertUnwindSafe; use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; use async_trait::async_trait; use futures::FutureExt; @@ -10,7 +9,7 @@ use fabro_core::context::Context as CoreContext; use fabro_core::error::{CoreError, HandlerErrorDetail, Result as CoreResult}; use fabro_core::handler::NodeHandler; use fabro_core::outcome::Outcome as CoreOutcome; -use fabro_core::retry::{BackoffPolicy, RetryPolicy as CoreRetryPolicy}; +use fabro_core::retry::RetryPolicy as CoreRetryPolicy; use super::graph::WorkflowGraph; use super::outcome::{wf_to_core_outcome, wf_to_core_status}; @@ -111,12 +110,7 @@ impl NodeHandler for WorkflowNodeHandler { let wf_policy = engine::build_retry_policy(gv_node, &gv_graph); CoreRetryPolicy { max_attempts: wf_policy.max_attempts, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(wf_policy.backoff.initial_delay_ms), - factor: wf_policy.backoff.backoff_factor, - max_delay: Duration::from_millis(wf_policy.backoff.max_delay_ms), - jitter: wf_policy.backoff.jitter, - }, + backoff: wf_policy.backoff, } } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs index 365e4334c..026761014 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs @@ -460,14 +460,6 @@ impl RunLifecycle for WorkflowLifecycle { _ => {} } - // Circuit breaker for loop_restart edges - if let Some(ref edge) = ctx.edge { - if edge.inner().loop_restart() { - // Check restart_failure_signatures limit - // (implemented in Phase 5 when full checkpoint resume is wired) - } - } - Ok(EdgeDecision::Continue) } diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 86e71519b..db832dec5 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -3,10 +3,11 @@ use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use chrono::Utc; use fabro_agent::Sandbox; +use fabro_util::backoff::BackoffPolicy; use futures::FutureExt; use rand::Rng; use tokio_util::sync::CancellationToken; @@ -67,60 +68,11 @@ struct LoopState { // --- Retry policy types --- -/// Configuration for exponential backoff between retry attempts. -#[derive(Debug, Clone)] -pub struct BackoffConfig { - pub initial_delay_ms: u64, - pub backoff_factor: f64, - pub max_delay_ms: u64, - pub jitter: bool, -} - -impl Default for BackoffConfig { - fn default() -> Self { - Self { - initial_delay_ms: 5_000, - backoff_factor: 2.0, - max_delay_ms: 60_000, - jitter: true, - } - } -} - -impl BackoffConfig { - /// Calculate delay for a given attempt (1-indexed). - #[must_use] - pub fn delay_for_attempt(&self, attempt: u32) -> std::time::Duration { - let exponent = attempt.saturating_sub(1); - let initial = f64::from(u32::try_from(self.initial_delay_ms).unwrap_or(u32::MAX)); - let max = f64::from(u32::try_from(self.max_delay_ms).unwrap_or(u32::MAX)); - let exp_i32 = i32::try_from(exponent).unwrap_or(i32::MAX); - let delay_f64 = initial * self.backoff_factor.powi(exp_i32); - let capped = delay_f64.min(max); - let final_ms = if self.jitter { - let mut rng = rand::thread_rng(); - let jitter_factor: f64 = rng.gen_range(0.5..1.5); - capped * jitter_factor - } else { - capped - }; - // f64 -> u64: clamp to non-negative, truncate via string-free path - let ms = if final_ms <= 0.0 { - 0u64 - } else if final_ms >= f64::from(u32::MAX) { - u64::from(u32::MAX) - } else { - final_ms as u64 - }; - std::time::Duration::from_millis(ms) - } -} - /// Retry policy for node execution. #[derive(Clone, Debug)] pub struct RetryPolicy { pub max_attempts: u32, - pub backoff: BackoffConfig, + pub backoff: BackoffPolicy, } impl RetryPolicy { @@ -129,19 +81,24 @@ impl RetryPolicy { pub fn none() -> Self { Self { max_attempts: 1, - backoff: BackoffConfig::default(), + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(5_000), + factor: 2.0, + max_delay: Duration::from_millis(60_000), + jitter: true, + }, } } /// Standard retry policy: 5 attempts, 5s initial, 2x factor. #[must_use] - pub const fn standard() -> Self { + pub fn standard() -> Self { Self { max_attempts: 5, - backoff: BackoffConfig { - initial_delay_ms: 5_000, - backoff_factor: 2.0, - max_delay_ms: 60_000, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(5_000), + factor: 2.0, + max_delay: Duration::from_millis(60_000), jitter: true, }, } @@ -149,13 +106,13 @@ impl RetryPolicy { /// Aggressive retry: 5 attempts, 500ms initial, 2x factor. #[must_use] - pub const fn aggressive() -> Self { + pub fn aggressive() -> Self { Self { max_attempts: 5, - backoff: BackoffConfig { - initial_delay_ms: 500, - backoff_factor: 2.0, - max_delay_ms: 60_000, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + factor: 2.0, + max_delay: Duration::from_millis(60_000), jitter: true, }, } @@ -163,13 +120,13 @@ impl RetryPolicy { /// Linear retry: 3 attempts, 500ms fixed delay. #[must_use] - pub const fn linear() -> Self { + pub fn linear() -> Self { Self { max_attempts: 3, - backoff: BackoffConfig { - initial_delay_ms: 500, - backoff_factor: 1.0, - max_delay_ms: 60_000, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + factor: 1.0, + max_delay: Duration::from_millis(60_000), jitter: true, }, } @@ -177,13 +134,13 @@ impl RetryPolicy { /// Patient retry: 3 attempts, 2000ms initial, 3x factor. #[must_use] - pub const fn patient() -> Self { + pub fn patient() -> Self { Self { max_attempts: 3, - backoff: BackoffConfig { - initial_delay_ms: 2000, - backoff_factor: 3.0, - max_delay_ms: 60_000, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(2000), + factor: 3.0, + max_delay: Duration::from_millis(60_000), jitter: true, }, } @@ -211,7 +168,12 @@ pub(crate) fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { let max_attempts = u32::try_from(max_retries + 1).unwrap_or(1).max(1); RetryPolicy { max_attempts, - backoff: BackoffConfig::default(), + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(5_000), + factor: 2.0, + max_delay: Duration::from_millis(60_000), + jitter: true, + }, } } @@ -1463,7 +1425,8 @@ impl WorkflowRunEngine { use fabro_core::state::RunState; use tokio_util::sync::CancellationToken; - let wf_graph = crate::core_adapter::WorkflowGraph(std::sync::Arc::new(graph.clone())); + let graph_arc = std::sync::Arc::new(graph.clone()); + let wf_graph = crate::core_adapter::WorkflowGraph(Arc::clone(&graph_arc)); // Build a shared EngineServices for the handler let shared_services = std::sync::Arc::new(EngineServices { @@ -1487,7 +1450,7 @@ impl WorkflowRunEngine { self.services.emitter.clone(), self.services.hook_runner.clone(), self.services.sandbox.clone(), - std::sync::Arc::new(graph.clone()), + graph_arc, config.run_dir.clone(), config.run_id.clone(), config.dry_run, @@ -2694,83 +2657,6 @@ mod tests { } } - // --- BackoffConfig tests --- - - #[test] - fn backoff_no_jitter_first_attempt() { - let config = BackoffConfig { - initial_delay_ms: 200, - backoff_factor: 2.0, - max_delay_ms: 60_000, - jitter: false, - }; - let delay = config.delay_for_attempt(1); - assert_eq!(delay.as_millis(), 200); - } - - #[test] - fn backoff_no_jitter_second_attempt() { - let config = BackoffConfig { - initial_delay_ms: 200, - backoff_factor: 2.0, - max_delay_ms: 60_000, - jitter: false, - }; - let delay = config.delay_for_attempt(2); - assert_eq!(delay.as_millis(), 400); - } - - #[test] - fn backoff_no_jitter_third_attempt() { - let config = BackoffConfig { - initial_delay_ms: 200, - backoff_factor: 2.0, - max_delay_ms: 60_000, - jitter: false, - }; - let delay = config.delay_for_attempt(3); - assert_eq!(delay.as_millis(), 800); - } - - #[test] - fn backoff_respects_max_delay() { - let config = BackoffConfig { - initial_delay_ms: 10_000, - backoff_factor: 10.0, - max_delay_ms: 30_000, - jitter: false, - }; - let delay = config.delay_for_attempt(5); - assert_eq!(delay.as_millis(), 30_000); - } - - #[test] - fn backoff_with_jitter_is_in_range() { - let config = BackoffConfig { - initial_delay_ms: 1000, - backoff_factor: 1.0, - max_delay_ms: 60_000, - jitter: true, - }; - let delay = config.delay_for_attempt(1); - // With jitter factor 0.5..1.5, delay should be 500..1500 - assert!(delay.as_millis() >= 500); - assert!(delay.as_millis() <= 1500); - } - - #[test] - fn backoff_linear_factor() { - let config = BackoffConfig { - initial_delay_ms: 500, - backoff_factor: 1.0, - max_delay_ms: 60_000, - jitter: false, - }; - assert_eq!(config.delay_for_attempt(1).as_millis(), 500); - assert_eq!(config.delay_for_attempt(2).as_millis(), 500); - assert_eq!(config.delay_for_attempt(3).as_millis(), 500); - } - // --- RetryPolicy preset tests --- #[test] @@ -2783,28 +2669,28 @@ mod tests { fn retry_policy_standard() { let policy = RetryPolicy::standard(); assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay_ms, 5_000); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(5_000)); } #[test] fn retry_policy_aggressive() { let policy = RetryPolicy::aggressive(); assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay_ms, 500); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(500)); } #[test] fn retry_policy_linear() { let policy = RetryPolicy::linear(); assert_eq!(policy.max_attempts, 3); - assert_eq!(policy.backoff.backoff_factor, 1.0); + assert_eq!(policy.backoff.factor, 1.0); } #[test] fn retry_policy_patient() { let policy = RetryPolicy::patient(); assert_eq!(policy.max_attempts, 3); - assert_eq!(policy.backoff.initial_delay_ms, 2000); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(2000)); } // --- build_retry_policy tests --- @@ -2848,7 +2734,7 @@ mod tests { let graph = Graph::new("test"); let policy = build_retry_policy(&node, &graph); assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay_ms, 500); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(500)); } #[test] @@ -2860,7 +2746,7 @@ mod tests { let policy = build_retry_policy(&node, &graph); assert_eq!(policy.max_attempts, 4); // 3 retries + 1 initial // Should use default backoff, not a preset's backoff - assert_eq!(policy.backoff.initial_delay_ms, 5_000); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(5_000)); } #[test]