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

camel_api/
error_handler.rs
