camel_api/
error_handler.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::{BoxProcessor, CamelError, Exchange, PipelineOutcome, SyncBoxProcessor};
5
6/// Camel-compatible header names for redelivery state.
7pub const HEADER_REDELIVERED: &str = "CamelRedelivered";
8pub const HEADER_REDELIVERY_COUNTER: &str = "CamelRedeliveryCounter";
9pub const HEADER_REDELIVERY_MAX_COUNTER: &str = "CamelRedeliveryMaxCounter";
10
11/// Redelivery policy with exponential backoff and optional jitter.
12#[derive(Debug, Clone)]
13pub struct RedeliveryPolicy {
14    pub max_attempts: u32,
15    pub initial_delay: Duration,
16    pub multiplier: f64,
17    pub max_delay: Duration,
18    pub jitter_factor: f64,
19}
20
21impl RedeliveryPolicy {
22    /// Create a new policy with default delays (100ms initial, 2x multiplier, 10s max, no jitter).
23    ///
24    /// Note: `max_attempts = 0` means no retries (immediate failure to DLC/handler).
25    /// Use `max_attempts > 0` to enable retry behavior.
26    pub fn new(max_attempts: u32) -> Self {
27        Self {
28            max_attempts,
29            initial_delay: Duration::from_millis(100),
30            multiplier: 2.0,
31            max_delay: Duration::from_secs(10),
32            jitter_factor: 0.0,
33        }
34    }
35
36    /// Override the initial delay before the first retry.
37    pub fn with_initial_delay(mut self, d: Duration) -> Self {
38        self.initial_delay = d;
39        self
40    }
41
42    /// Override the backoff multiplier applied after each attempt.
43    pub fn with_multiplier(mut self, m: f64) -> Self {
44        self.multiplier = m;
45        self
46    }
47
48    /// Cap the maximum delay between retries.
49    pub fn with_max_delay(mut self, d: Duration) -> Self {
50        self.max_delay = d;
51        self
52    }
53
54    /// Set jitter factor (0.0 = no jitter, 0.2 = ±20% randomization).
55    ///
56    /// Recommended values: 0.1-0.3 (10-30%) for most use cases.
57    /// Helps prevent thundering herd problems in distributed systems
58    /// by adding randomization to retry timing.
59    pub fn with_jitter(mut self, j: f64) -> Self {
60        self.jitter_factor = j.clamp(0.0, 1.0);
61        self
62    }
63
64    /// Compute the sleep duration before retry attempt N (0-indexed) with jitter applied.
65    pub fn delay_for(&self, attempt: u32) -> Duration {
66        let base_ms = self.initial_delay.as_millis() as f64 * self.multiplier.powi(attempt as i32);
67        let capped_ms = base_ms.min(self.max_delay.as_millis() as f64);
68
69        if self.jitter_factor > 0.0 {
70            let jitter = capped_ms * self.jitter_factor * (rand::random::<f64>() * 2.0 - 1.0);
71            Duration::from_millis((capped_ms + jitter).max(0.0) as u64)
72        } else {
73            Duration::from_millis(capped_ms as u64)
74        }
75    }
76}
77
78/// Disposition for an exception handler or catch clause (ADR-0019).
79///
80/// # YAML casing
81///
82/// serde uses `lowercase` casing: `handled`, `propagate`, `continued`.
83///
84/// # Default divergence
85///
86/// `ExceptionDisposition::default()` returns `Propagate` (first variant per `#[derive(Default)]`),
87/// but the YAML + builder layers default to `Handled` for doTry catch clauses
88/// (via `default_handled_disposition()` in `camel-dsl/src/route_ast.rs` and the `DoCatchBuilder::do_catch_exception` constructor).
89/// Direct struct construction should explicitly set disposition rather than relying on `Default`.
90#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
91#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Deserialize)]
92#[serde(rename_all = "lowercase")]
93pub enum ExceptionDisposition {
94    /// After retry exhaustion / on_steps / DLC, re-throw to upstream.
95    #[default]
96    Propagate,
97    /// Suppress re-throw. The handler's exchange is the final result.
98    Handled,
99    /// Clear the error. The pipeline continues to the NEXT step.
100    Continued,
101}
102
103/// Opaque identifier for a matched ExceptionPolicy within a RouteErrorHandler.
104/// Index into the policies Vec — valid only within the handler that created it.
105/// If the policies vec is reordered or filtered, indices become stale.
106/// Safe because policies are immutable after handler construction.
107#[derive(Clone, Copy, Debug, PartialEq, Eq)]
108pub struct PolicyId(pub usize);
109
110/// Result of a single retry attempt by `RouteErrorHandler::retry_step`.
111#[derive(Debug)]
112pub enum RetryOutcome {
113    /// Retry succeeded; pipeline continues with this Exchange.
114    Recovered(Exchange),
115    /// Retry attempt returned `Stopped(ex)`. Bypasses `handle_step` —
116    /// Stop is successful control flow, not exhausted error handling.
117    /// Maps to `PipelineOutcome::Stopped(ex)` in `run_steps`.
118    Stopped(Exchange),
119    /// All retry attempts exhausted. Handler decides via `handle_step`.
120    Exhausted {
121        exchange: Exchange,
122        error: CamelError,
123        policy: Option<PolicyId>,
124    },
125}
126
127/// Object-safe retry abstraction. Unifies `BoxProcessor` and `OutcomeSegment`
128/// for `RouteErrorHandler::retry_step`. See ADR-0024 amendment (Phase 4).
129///
130/// `invoke` returns `PipelineOutcome` (not Tower `Result`), preserving
131/// `Stopped(ex)` state across retry attempts.
132pub trait RetryableStep: Send {
133    fn invoke<'a>(
134        &'a mut self,
135        exchange: Exchange,
136    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = PipelineOutcome> + Send + 'a>>;
137}
138
139/// BoxProcessor adapter: preserves readiness-error routing by calling
140/// `ServiceExt::ready().await` before `Service::call()`. Readiness
141/// failures (e.g. channel closed, circuit breaker open) become
142/// `PipelineOutcome::Failed`, NOT panic or silent skip.
143impl RetryableStep for crate::BoxProcessor {
144    fn invoke<'a>(
145        &'a mut self,
146        exchange: Exchange,
147    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = PipelineOutcome> + Send + 'a>> {
148        use tower::ServiceExt;
149        Box::pin(async move {
150            match self.ready().await {
151                Ok(ready_svc) => match tower::Service::call(ready_svc, exchange).await {
152                    Ok(ex) => PipelineOutcome::Completed(ex),
153                    Err(err) => PipelineOutcome::Failed(err),
154                },
155                Err(err) => PipelineOutcome::Failed(err),
156            }
157        })
158    }
159}
160
161/// Result of Phase 2 (handle) of step error handling.
162/// Handled and Continued MUST have Exchange.error cleared.
163pub enum StepDisposition {
164    Propagate(CamelError),
165    Handled(Exchange),
166    Continued(Exchange),
167}
168
169/// Identifies which boundary gate produced an infrastructure error.
170#[derive(Debug, Clone, Copy, PartialEq, Eq)]
171pub enum BoundaryKind {
172    Security,
173    CircuitBreaker,
174    Readiness,
175}
176
177/// A rule that matches specific errors and defines retry + redirect behaviour.
178pub struct ExceptionPolicy {
179    /// Predicate: returns `true` if this policy applies to the given error.
180    pub matches: Arc<dyn Fn(&CamelError) -> bool + Send + Sync>,
181    /// Optional retry configuration; if absent, no retries are attempted.
182    pub retry: Option<RedeliveryPolicy>,
183    /// Optional URI of a specific endpoint to route failed exchanges to.
184    pub handled_by: Option<String>,
185    /// Optional custom pipeline executed when this policy triggers.
186    pub on_steps: Option<SyncBoxProcessor>,
187    /// What to do after this policy's handler runs.
188    pub disposition: ExceptionDisposition,
189}
190
191impl ExceptionPolicy {
192    /// Create a new policy that matches errors using the given predicate.
193    pub fn new(matches: impl Fn(&CamelError) -> bool + Send + Sync + 'static) -> Self {
194        Self {
195            matches: Arc::new(matches),
196            retry: None,
197            handled_by: None,
198            on_steps: None,
199            disposition: ExceptionDisposition::Propagate,
200        }
201    }
202}
203
204impl Clone for ExceptionPolicy {
205    fn clone(&self) -> Self {
206        Self {
207            matches: Arc::clone(&self.matches),
208            retry: self.retry.clone(),
209            handled_by: self.handled_by.clone(),
210            on_steps: self.on_steps.clone(),
211            disposition: self.disposition,
212        }
213    }
214}
215
216/// Full error handler configuration: Dead Letter Channel URI and per-exception policies.
217#[derive(Clone)]
218pub struct ErrorHandlerConfig {
219    /// URI of the Dead Letter Channel endpoint (None = log only).
220    pub dlc_uri: Option<String>,
221    /// Per-exception policies evaluated in order; first match wins.
222    pub policies: Vec<ExceptionPolicy>,
223    /// When true, restore the original (pre-route, pre-mutation) Message
224    /// (body and headers) before forwarding to the DLC/handler.
225    pub use_original_message: bool,
226}
227
228impl ErrorHandlerConfig {
229    /// Log-only error handler: errors are logged but not forwarded anywhere.
230    pub fn log_only() -> Self {
231        Self {
232            dlc_uri: None,
233            policies: Vec::new(),
234            use_original_message: false,
235        }
236    }
237
238    /// Dead Letter Channel: failed exchanges are forwarded to the given URI.
239    pub fn dead_letter_channel(uri: impl Into<String>) -> Self {
240        Self {
241            dlc_uri: Some(uri.into()),
242            policies: Vec::new(),
243            use_original_message: false,
244        }
245    }
246
247    /// Start building an `ExceptionPolicy` attached to this config.
248    pub fn on_exception(
249        self,
250        matches: impl Fn(&CamelError) -> bool + Send + Sync + 'static,
251    ) -> ExceptionPolicyBuilder {
252        ExceptionPolicyBuilder {
253            config: self,
254            policy: ExceptionPolicy::new(matches),
255        }
256    }
257}
258
259/// Builder for a single [`ExceptionPolicy`] attached to an [`ErrorHandlerConfig`].
260pub struct ExceptionPolicyBuilder {
261    config: ErrorHandlerConfig,
262    policy: ExceptionPolicy,
263}
264
265impl ExceptionPolicyBuilder {
266    /// Configure retry with the given maximum number of attempts (exponential backoff defaults).
267    pub fn retry(mut self, max_attempts: u32) -> Self {
268        self.policy.retry = Some(RedeliveryPolicy::new(max_attempts));
269        self
270    }
271
272    /// Override backoff parameters for the retry (call after `.retry()`).
273    pub fn with_backoff(mut self, initial: Duration, multiplier: f64, max: Duration) -> Self {
274        if let Some(ref mut p) = self.policy.retry {
275            p.initial_delay = initial;
276            p.multiplier = multiplier;
277            p.max_delay = max;
278        }
279        self
280    }
281
282    /// Set jitter factor for retry delays (call after `.retry()`).
283    /// Valid range: 0.0 (no jitter) to 1.0 (±100% randomization).
284    pub fn with_jitter(mut self, jitter_factor: f64) -> Self {
285        if let Some(ref mut p) = self.policy.retry {
286            p.jitter_factor = jitter_factor.clamp(0.0, 1.0);
287        }
288        self
289    }
290
291    /// Route failed exchanges matching this policy to the given URI instead of the DLC.
292    pub fn handled_by(mut self, uri: impl Into<String>) -> Self {
293        self.policy.handled_by = Some(uri.into());
294        self
295    }
296
297    /// Attach a custom pipeline to execute when this policy triggers.
298    pub fn on_steps(mut self, pipeline: BoxProcessor) -> Self {
299        self.policy.on_steps = Some(SyncBoxProcessor::new(pipeline));
300        self
301    }
302
303    /// Mark the exception as handled (suppresses re-throw to upstream).
304    pub fn handled(mut self, handled: bool) -> Self {
305        self.policy.disposition = if handled {
306            ExceptionDisposition::Handled
307        } else {
308            ExceptionDisposition::Propagate
309        };
310        self
311    }
312
313    /// Mark the exception as continued (clear error, pipeline continues to next step).
314    pub fn continued(mut self, continued: bool) -> Self {
315        self.policy.disposition = if continued {
316            ExceptionDisposition::Continued
317        } else {
318            ExceptionDisposition::Propagate
319        };
320        self
321    }
322
323    /// Explicitly set disposition to Propagate (default, no-op).
324    pub fn propagate(mut self) -> Self {
325        self.policy.disposition = ExceptionDisposition::Propagate;
326        self
327    }
328
329    /// Finish this policy and return the updated config.
330    pub fn build(mut self) -> ErrorHandlerConfig {
331        self.config.policies.push(self.policy);
332        self.config
333    }
334}
335
336// Backwards compatibility alias
337#[deprecated(since = "0.1.0", note = "Use `RedeliveryPolicy` instead")]
338pub type ExponentialBackoff = RedeliveryPolicy;
339
340#[cfg(test)]
341mod retryable_step_tests {
342    use super::*;
343    use crate::{BoxProcessor, CamelError, Exchange, Message, PipelineOutcome};
344    use std::future::Future;
345    use std::pin::Pin;
346    use std::sync::Arc;
347    use std::sync::atomic::{AtomicUsize, Ordering};
348
349    struct CountingProcessor {
350        call_count: Arc<AtomicUsize>,
351        succeed: bool,
352    }
353
354    impl tower::Service<Exchange> for CountingProcessor {
355        type Response = Exchange;
356        type Error = CamelError;
357        type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
358
359        fn poll_ready(
360            &mut self,
361            _cx: &mut std::task::Context<'_>,
362        ) -> std::task::Poll<Result<(), Self::Error>> {
363            std::task::Poll::Ready(Ok(()))
364        }
365
366        fn call(&mut self, exchange: Exchange) -> Self::Future {
367            let count = self.call_count.clone();
368            let succeed = self.succeed;
369            Box::pin(async move {
370                count.fetch_add(1, Ordering::SeqCst);
371                if succeed {
372                    Ok(exchange)
373                } else {
374                    Err(CamelError::ProcessorError("fail".into()))
375                }
376            })
377        }
378    }
379
380    impl Clone for CountingProcessor {
381        fn clone(&self) -> Self {
382            Self {
383                call_count: self.call_count.clone(),
384                succeed: self.succeed,
385            }
386        }
387    }
388
389    #[tokio::test]
390    async fn boxprocessor_adapter_maps_ok_to_completed() {
391        let count = Arc::new(AtomicUsize::new(0));
392        let processor = CountingProcessor {
393            call_count: count.clone(),
394            succeed: true,
395        };
396        let bp: BoxProcessor = BoxProcessor::new(processor);
397        let mut retryable: Box<dyn RetryableStep> = Box::new(bp);
398        let ex = Exchange::new(Message::new("hello"));
399        let outcome = retryable.invoke(ex).await;
400        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
401        assert_eq!(count.load(Ordering::SeqCst), 1);
402    }
403
404    #[tokio::test]
405    async fn boxprocessor_adapter_maps_err_to_failed() {
406        let processor = CountingProcessor {
407            call_count: Arc::new(AtomicUsize::new(0)),
408            succeed: false,
409        };
410        let bp: BoxProcessor = BoxProcessor::new(processor);
411        let mut retryable: Box<dyn RetryableStep> = Box::new(bp);
412        let ex = Exchange::new(Message::new("hello"));
413        let outcome = retryable.invoke(ex).await;
414        assert!(matches!(outcome, PipelineOutcome::Failed(_)));
415    }
416
417    #[tokio::test]
418    async fn boxprocessor_readiness_error_propagates_to_failed() {
419        struct AlwaysNotReady;
420
421        impl tower::Service<Exchange> for AlwaysNotReady {
422            type Response = Exchange;
423            type Error = CamelError;
424            type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
425
426            fn poll_ready(
427                &mut self,
428                _cx: &mut std::task::Context<'_>,
429            ) -> std::task::Poll<Result<(), Self::Error>> {
430                std::task::Poll::Ready(Err(CamelError::ProcessorError(
431                    "readiness failed: consumer closed".into(),
432                )))
433            }
434
435            fn call(&mut self, _ex: Exchange) -> Self::Future {
436                Box::pin(async {
437                    unreachable!("call() must not be reached when poll_ready errors")
438                })
439            }
440        }
441
442        impl Clone for AlwaysNotReady {
443            fn clone(&self) -> Self {
444                AlwaysNotReady
445            }
446        }
447
448        let bp: BoxProcessor = BoxProcessor::new(AlwaysNotReady);
449        let mut retryable: Box<dyn RetryableStep> = Box::new(bp);
450        let ex = Exchange::new(Message::new("hello"));
451        let outcome = retryable.invoke(ex).await;
452        match outcome {
453            PipelineOutcome::Failed(err) => {
454                let msg = err.to_string();
455                assert!(
456                    msg.contains("readiness") || msg.contains("consumer"),
457                    "readiness error message should be preserved, got: {msg}"
458                );
459            }
460            other => panic!(
461                "readiness failure must map to PipelineOutcome::Failed, got {:?}",
462                other
463            ),
464        }
465    }
466}
467
468#[cfg(test)]
469mod tests {
470    use super::*;
471    use crate::CamelError;
472    use std::time::Duration;
473
474    #[test]
475    fn test_redelivery_policy_defaults() {
476        let p = RedeliveryPolicy::new(3);
477        assert_eq!(p.max_attempts, 3);
478        assert_eq!(p.initial_delay, Duration::from_millis(100));
479        assert_eq!(p.multiplier, 2.0);
480        assert_eq!(p.max_delay, Duration::from_secs(10));
481        assert_eq!(p.jitter_factor, 0.0);
482    }
483
484    #[test]
485    fn test_exception_policy_matches() {
486        let policy = ExceptionPolicy::new(|e| matches!(e, CamelError::ProcessorError(_)));
487        assert!((policy.matches)(&CamelError::ProcessorError("oops".into())));
488        assert!(!(policy.matches)(&CamelError::Io("io".into())));
489    }
490
491    #[test]
492    fn test_error_handler_config_log_only() {
493        let config = ErrorHandlerConfig::log_only();
494        assert!(config.dlc_uri.is_none());
495        assert!(config.policies.is_empty());
496    }
497
498    #[test]
499    fn test_error_handler_config_dlc() {
500        let config = ErrorHandlerConfig::dead_letter_channel("log:dlc");
501        assert_eq!(config.dlc_uri.as_deref(), Some("log:dlc"));
502    }
503
504    #[test]
505    fn test_error_handler_config_with_policy() {
506        let config = ErrorHandlerConfig::dead_letter_channel("log:dlc")
507            .on_exception(|e| matches!(e, CamelError::Io(_)))
508            .retry(2)
509            .handled_by("log:io-errors")
510            .build();
511        assert_eq!(config.policies.len(), 1);
512        let p = &config.policies[0];
513        assert!(p.retry.is_some());
514        assert_eq!(p.retry.as_ref().unwrap().max_attempts, 2);
515        assert_eq!(p.handled_by.as_deref(), Some("log:io-errors"));
516    }
517
518    #[test]
519    fn test_jitter_applies_randomness() {
520        let policy = RedeliveryPolicy::new(3)
521            .with_initial_delay(Duration::from_millis(100))
522            .with_jitter(0.5);
523
524        let mut delays = std::collections::HashSet::new();
525        for _ in 0..10 {
526            delays.insert(policy.delay_for(0));
527        }
528
529        assert!(delays.len() > 1, "jitter should produce varying delays");
530    }
531
532    #[test]
533    fn test_jitter_stays_within_bounds() {
534        let policy = RedeliveryPolicy::new(3)
535            .with_initial_delay(Duration::from_millis(100))
536            .with_jitter(0.5);
537
538        for _ in 0..100 {
539            let delay = policy.delay_for(0);
540            assert!(
541                delay >= Duration::from_millis(50),
542                "delay too low: {:?}",
543                delay
544            );
545            assert!(
546                delay <= Duration::from_millis(150),
547                "delay too high: {:?}",
548                delay
549            );
550        }
551    }
552
553    #[test]
554    fn test_max_attempts_zero_means_no_retries() {
555        let policy = RedeliveryPolicy::new(0);
556        assert_eq!(policy.max_attempts, 0);
557    }
558
559    #[test]
560    fn test_jitter_zero_produces_exact_delay() {
561        let policy = RedeliveryPolicy::new(3)
562            .with_initial_delay(Duration::from_millis(100))
563            .with_jitter(0.0);
564
565        for _ in 0..10 {
566            let delay = policy.delay_for(0);
567            assert_eq!(delay, Duration::from_millis(100));
568        }
569    }
570
571    #[test]
572    fn test_jitter_one_produces_wide_range() {
573        let policy = RedeliveryPolicy::new(3)
574            .with_initial_delay(Duration::from_millis(100))
575            .with_jitter(1.0);
576
577        for _ in 0..100 {
578            let delay = policy.delay_for(0);
579            assert!(
580                delay >= Duration::from_millis(0),
581                "delay should be >= 0, got {:?}",
582                delay
583            );
584            assert!(
585                delay <= Duration::from_millis(200),
586                "delay should be <= 200ms, got {:?}",
587                delay
588            );
589        }
590    }
591
592    #[test]
593    fn test_redelivery_policy_builder_methods_apply_values() {
594        let p = RedeliveryPolicy::new(5)
595            .with_initial_delay(Duration::from_millis(250))
596            .with_multiplier(3.0)
597            .with_max_delay(Duration::from_secs(2))
598            .with_jitter(2.0);
599
600        assert_eq!(p.initial_delay, Duration::from_millis(250));
601        assert_eq!(p.multiplier, 3.0);
602        assert_eq!(p.max_delay, Duration::from_secs(2));
603        assert_eq!(p.jitter_factor, 1.0);
604    }
605
606    #[test]
607    fn test_with_jitter_clamps_low_bound() {
608        let p = RedeliveryPolicy::new(1).with_jitter(-0.2);
609        assert_eq!(p.jitter_factor, 0.0);
610    }
611
612    #[test]
613    fn test_delay_for_exponential_growth_and_cap() {
614        let p = RedeliveryPolicy::new(3)
615            .with_initial_delay(Duration::from_millis(100))
616            .with_multiplier(2.0)
617            .with_max_delay(Duration::from_millis(250));
618
619        assert_eq!(p.delay_for(0), Duration::from_millis(100));
620        assert_eq!(p.delay_for(1), Duration::from_millis(200));
621        assert_eq!(p.delay_for(2), Duration::from_millis(250));
622        assert_eq!(p.delay_for(20), Duration::from_millis(250));
623    }
624
625    #[test]
626    fn test_exception_policy_builder_backoff_and_jitter() {
627        let config = ErrorHandlerConfig::log_only()
628            .on_exception(|e| matches!(e, CamelError::Io(_)))
629            .retry(4)
630            .with_backoff(Duration::from_millis(10), 1.5, Duration::from_millis(40))
631            .with_jitter(1.5)
632            .build();
633
634        let retry = config.policies[0].retry.as_ref().unwrap();
635        assert_eq!(retry.max_attempts, 4);
636        assert_eq!(retry.initial_delay, Duration::from_millis(10));
637        assert_eq!(retry.multiplier, 1.5);
638        assert_eq!(retry.max_delay, Duration::from_millis(40));
639        assert_eq!(retry.jitter_factor, 1.0);
640    }
641
642    #[test]
643    fn test_exception_policy_builder_no_retry_ignores_backoff_and_jitter() {
644        let config = ErrorHandlerConfig::log_only()
645            .on_exception(|_| true)
646            .with_backoff(Duration::from_secs(1), 9.0, Duration::from_secs(2))
647            .with_jitter(0.8)
648            .build();
649
650        assert!(config.policies[0].retry.is_none());
651    }
652
653    #[test]
654    fn test_exception_policy_clone_preserves_behavior_and_fields() {
655        let policy = ExceptionPolicy::new(|e| matches!(e, CamelError::RouteError(_)));
656        let mut configured = policy;
657        configured.retry = Some(RedeliveryPolicy::new(2));
658        configured.handled_by = Some("log:route-errors".to_string());
659
660        let cloned = configured.clone();
661        assert!((cloned.matches)(&CamelError::RouteError("x".into())));
662        assert_eq!(cloned.retry.as_ref().unwrap().max_attempts, 2);
663        assert_eq!(cloned.handled_by.as_deref(), Some("log:route-errors"));
664    }
665
666    #[test]
667    fn test_delay_for_respects_max_delay_with_jitter() {
668        let policy = RedeliveryPolicy::new(5)
669            .with_initial_delay(Duration::from_millis(200))
670            .with_multiplier(10.0)
671            .with_max_delay(Duration::from_millis(500))
672            .with_jitter(0.2);
673
674        for _ in 0..30 {
675            let delay = policy.delay_for(4);
676            assert!(delay <= Duration::from_millis(600));
677            assert!(delay >= Duration::from_millis(400));
678        }
679    }
680
681    #[test]
682    fn test_exception_policy_builder_keeps_dlc_and_policy_order() {
683        let config = ErrorHandlerConfig::dead_letter_channel("log:dlc")
684            .on_exception(|e| matches!(e, CamelError::Io(_)))
685            .retry(1)
686            .build()
687            .on_exception(|e| matches!(e, CamelError::RouteError(_)))
688            .handled_by("log:routes")
689            .build();
690
691        assert_eq!(config.dlc_uri.as_deref(), Some("log:dlc"));
692        assert_eq!(config.policies.len(), 2);
693        assert!((config.policies[0].matches)(&CamelError::Io("x".into())));
694        assert!((config.policies[1].matches)(&CamelError::RouteError(
695            "x".into()
696        )));
697    }
698
699    #[test]
700    fn test_backoff_without_retry_does_not_create_retry_config() {
701        let config = ErrorHandlerConfig::log_only()
702            .on_exception(|_| true)
703            .with_backoff(Duration::from_millis(1), 3.0, Duration::from_millis(9))
704            .build();
705
706        assert!(config.policies[0].retry.is_none());
707    }
708
709    #[test]
710    fn test_exception_disposition_default_is_propagate() {
711        assert_eq!(
712            ExceptionDisposition::default(),
713            ExceptionDisposition::Propagate
714        );
715    }
716
717    #[test]
718    fn test_exception_policy_new_has_propagate_disposition() {
719        let p = ExceptionPolicy::new(|_| true);
720        assert_eq!(p.disposition, ExceptionDisposition::Propagate);
721    }
722
723    #[test]
724    fn test_policy_id_equality() {
725        assert_eq!(PolicyId(0), PolicyId(0));
726        assert_ne!(PolicyId(0), PolicyId(1));
727    }
728
729    #[test]
730    fn test_builder_continued_sets_disposition() {
731        let cfg = ErrorHandlerConfig::log_only()
732            .on_exception(|_| true)
733            .continued(true)
734            .build();
735        assert_eq!(cfg.policies[0].disposition, ExceptionDisposition::Continued);
736    }
737
738    #[test]
739    fn test_builder_propagate_sets_disposition() {
740        let cfg = ErrorHandlerConfig::log_only()
741            .on_exception(|_| true)
742            .propagate()
743            .build();
744        assert_eq!(cfg.policies[0].disposition, ExceptionDisposition::Propagate);
745    }
746
747    #[test]
748    fn test_builder_handled_true_still_works() {
749        let cfg = ErrorHandlerConfig::log_only()
750            .on_exception(|_| true)
751            .handled(true)
752            .build();
753        assert_eq!(cfg.policies[0].disposition, ExceptionDisposition::Handled);
754    }
755}
camel_api/error_handler.rs

camel_api/
error_handler.rs
