camel_processor/
do_try.rs

1//! ## Stop semantics (ADR-0025)
2//!
3//! This segment implements `OutcomePipeline` and propagates `PipelineOutcome::Stopped(ex)` with the exchange state intact (including mutations made inside the segment body before Stop fired). See ADR-0025 §3 (stopped-exchange-state-preservation invariant).
4//!
5//! Catch-block failure envelope (bd rc-zgbqq): if a catch body fails, the catch error stays the main error in every disposition; the original error lives in the `warn` record ("do_try catch block failed; catch error supersedes original", fields `original_error` and `catch_error`) and, when a span is active, in the span event.
6//! See the error-handler spec, requirement "delegate failure propagates the original error" (that rule covers `handled_by` delegates, not catch bodies).
7
8use camel_api::error_handler::ExceptionDisposition;
9use camel_api::exchange::PROPERTY_EXCEPTION_HANDLED;
10use camel_api::{BoxProcessor, CamelError, Exchange, FilterPredicate};
11use tower::Service;
12use tower::ServiceExt;
13
14use crate::error_handler::record_span_error;
15
16/// Matcher for a `doCatch` clause.
17#[derive(Clone)]
18pub enum CatchMatcher {
19    /// Match by CamelError variant name (e.g. ["ProcessorError", "Io"]).
20    /// `"*"` matches any variant — equivalent to Camel's `doCatch(Throwable.class)`.
21    ByVariant(Vec<String>),
22    /// Match by predicate over the Exchange.
23    Predicate(FilterPredicate),
24}
25
26impl CatchMatcher {
27    /// Returns `true` if this matcher matches the given error and exchange.
28    pub fn matches(&self, err: &CamelError, ex: &Exchange) -> bool {
29        match self {
30            CatchMatcher::ByVariant(names) => {
31                if names.iter().any(|n| n == "*") {
32                    return true;
33                }
34                names.iter().any(|n| n == err.variant_name())
35            }
36            CatchMatcher::Predicate(p) => p(ex),
37        }
38    }
39}
40
41/// A single `doCatch` clause.
42#[derive(Clone)]
43pub struct CatchClause {
44    /// The main matcher (variant-name list or predicate).
45    pub matcher: CatchMatcher,
46    /// Optional sub-predicate evaluated AFTER the main matcher passes.
47    pub on_when: Option<FilterPredicate>,
48    /// Sub-pipeline executed when the clause matches.
49    pub steps: Vec<BoxProcessor>,
50    /// ADR-0019 disposition: Handled (default), Propagate, or Continued (rejected at parse time).
51    /// In YAML, use lowercase: `handled`, `propagate`, `continued`.
52    pub disposition: ExceptionDisposition,
53}
54
55/// The `doTry` processor. Wrap with `BoxProcessor::new(DoTryService::new(...))`.
56#[derive(Clone)]
57pub struct DoTryService {
58    /// Steps in the try block.
59    pub try_steps: Vec<BoxProcessor>,
60    /// Catch clauses evaluated first-match-wins.
61    pub catch_clauses: Vec<CatchClause>,
62    /// Steps in the finally block (empty = no finally).
63    pub finally_steps: Vec<BoxProcessor>,
64    /// Optional onWhen predicate for finally.
65    pub finally_on_when: Option<FilterPredicate>,
66}
67
68impl DoTryService {
69    /// Create a new `DoTryService` with the given try steps.
70    pub fn new(try_steps: Vec<BoxProcessor>) -> Self {
71        Self {
72            try_steps,
73            catch_clauses: Vec::new(),
74            finally_steps: Vec::new(),
75            finally_on_when: None,
76        }
77    }
78
79    /// Full constructor used by the compile pipeline (Task 10b control_flow.rs).
80    /// Builder API (Task 8) constructs via `new()` + field mutation.
81    pub fn with_catch_and_finally(
82        try_steps: Vec<BoxProcessor>,
83        catch_clauses: Vec<CatchClause>,
84        finally_steps: Vec<BoxProcessor>,
85        finally_on_when: Option<FilterPredicate>,
86    ) -> Self {
87        Self {
88            try_steps,
89            catch_clauses,
90            finally_steps,
91            finally_on_when,
92        }
93    }
94}
95
96/// Run a sequence of steps, preserving the last exchange state on error.
97/// Returns `Err(Box<(last_ex, err)>)` so DoTry can populate exception properties.
98async fn run_pipeline(
99    steps: Vec<BoxProcessor>,
100    mut ex: Exchange,
101) -> Result<Exchange, Box<(Exchange, CamelError)>> {
102    for mut svc in steps {
103        match svc.ready().await {
104            Ok(ready) => {
105                let snapshot = ex.clone();
106                match ready.call(ex).await {
107                    Ok(new_ex) => ex = new_ex,
108                    Err(err) => return Err(Box::new((snapshot, err))),
109                }
110            }
111            Err(err) => return Err(Box::new((ex, err))),
112        }
113    }
114    Ok(ex)
115}
116
117/// Run the finally block. Camel parity for finally-throws:
118/// - If finally succeeds (or is skipped): `Completed` with its exchange.
119/// - If finally throws AND there was a previous error: `Restore` (caller
120///   logs and restores the previous error).
121/// - If finally throws AND no previous error: `NoPreviousFail` (caller logs
122///   and propagates finally_err).
123///
124/// Logging lives at the CALLERS so the restore record's field names can
125/// match the calling flow (`previous_error` vs `catch_error`, bd rc-zgbqq).
126async fn run_finally(
127    finally_steps: Vec<BoxProcessor>,
128    finally_on_when: Option<FilterPredicate>,
129    ex: Exchange,
130    previous_err: Option<CamelError>,
131) -> FinallyOutcome {
132    if finally_steps.is_empty() {
133        return FinallyOutcome::Completed(ex);
134    }
135    if let Some(on_when) = &finally_on_when
136        && !on_when(&ex)
137    {
138        return FinallyOutcome::Completed(ex);
139    }
140    match run_pipeline(finally_steps, ex).await {
141        Ok(ex) => FinallyOutcome::Completed(ex),
142        Err(failed) => {
143            let (_, finally_err) = *failed;
144            match previous_err {
145                Some(prev) => FinallyOutcome::Restore {
146                    previous: prev,
147                    finally_err,
148                },
149                None => FinallyOutcome::NoPreviousFail(finally_err),
150            }
151        }
152    }
153}
154
155/// Result of `run_finally`. Tower-local to this module — unrelated to the
156/// identically-named enum in `do_try_segment.rs`.
157// The variant shape is fixed by the sealed do_try ruling (bd rc-zgbqq);
158// Exchange simply dwarfs the error payloads, so allow the lint instead of
159// boxing and drifting from the specified envelope.
160#[allow(clippy::large_enum_variant)]
161enum FinallyOutcome {
162    /// Finally ran (or was skipped) successfully; carries its exchange.
163    Completed(Exchange),
164    /// Finally threw with no previous error; the finally error wins.
165    NoPreviousFail(CamelError),
166    /// Finally threw with a previous error; the previous error is restored.
167    Restore {
168        previous: CamelError,
169        finally_err: CamelError,
170    },
171}
172
173impl tower::Service<Exchange> for DoTryService {
174    type Response = Exchange;
175    type Error = CamelError;
176    type Future = std::pin::Pin<
177        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
178    >;
179
180    fn poll_ready(
181        &mut self,
182        _cx: &mut std::task::Context<'_>,
183    ) -> std::task::Poll<Result<(), Self::Error>> {
184        std::task::Poll::Ready(Ok(()))
185    }
186
187    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
188        // Clear stale CamelExceptionHandled marker from prior handlers in the same route.
189        // Exchange::clear_error() does NOT touch HANDLED, so direct map access is required.
190        exchange.properties.remove(PROPERTY_EXCEPTION_HANDLED);
191
192        let try_steps = self.try_steps.clone();
193        let catch_clauses = self.catch_clauses.clone();
194        let finally_steps = self.finally_steps.clone();
195        let finally_on_when = self.finally_on_when.clone();
196
197        Box::pin(async move {
198            let try_result = run_pipeline(try_steps, exchange).await;
199            match try_result {
200                Ok(ex) => match run_finally(finally_steps, finally_on_when, ex, None).await {
201                    FinallyOutcome::Completed(ex) => Ok(ex),
202                    FinallyOutcome::NoPreviousFail(fin) => {
203                        tracing::warn!(error = %fin, "doFinally threw");
204                        Err(fin)
205                    }
206                    // Unreachable: the try-Ok flow passes no previous error.
207                    FinallyOutcome::Restore { previous, .. } => Err(previous),
208                },
209                Err(failed) => {
210                    let (failed_ex, original_err) = *failed;
211                    let mut ex = failed_ex;
212                    ex.set_error(original_err.clone());
213
214                    for clause in catch_clauses {
215                        let CatchClause {
216                            matcher,
217                            on_when,
218                            steps,
219                            disposition,
220                        } = clause;
221                        if !matcher.matches(&original_err, &ex) {
222                            continue;
223                        }
224                        if let Some(ref on_when) = on_when
225                            && !on_when(&ex)
226                        {
227                            continue;
228                        }
229
230                        let catch_result = run_pipeline(steps, ex.clone()).await;
231
232                        return match catch_result {
233                            Ok(ok_ex) => {
234                                // Determine previous-error threading based on disposition.
235                                // IMPORTANT: do NOT call handle_error() before run_finally() —
236                                // handle_error() calls clear_error() which removes
237                                // PROPERTY_EXCEPTION_MESSAGE/KIND/CAUGHT, preventing finally
238                                // steps from inspecting the caught exception.
239                                //
240                                // disposition semantics (ADR-0019 strict):
241                                //   Handled    -> catch output is final, no propagation
242                                //   Propagate  -> catch ran for side-effects, original propagates
243                                //   Continued  -> rejected at parse time (defensive: treat as
244                                //                 Propagate + log if we ever reach runtime)
245                                let prev = match disposition {
246                                    ExceptionDisposition::Handled => None,
247                                    ExceptionDisposition::Continued => {
248                                        tracing::warn!(
249                                            "ExceptionDisposition::Continued reached doTry runtime; \
250                                             treating as Propagate. Should have been rejected at parse time."
251                                        );
252                                        Some(original_err.clone())
253                                    }
254                                    // Propagate and any future variant thread the original error.
255                                    _ => Some(original_err.clone()),
256                                };
257                                let mut ex = match run_finally(
258                                    finally_steps.clone(),
259                                    finally_on_when.clone(),
260                                    ok_ex,
261                                    prev,
262                                )
263                                .await
264                                {
265                                    FinallyOutcome::Completed(ex) => ex,
266                                    FinallyOutcome::NoPreviousFail(fin) => {
267                                        tracing::warn!(error = %fin, "doFinally threw");
268                                        return Err(fin);
269                                    }
270                                    FinallyOutcome::Restore {
271                                        previous,
272                                        finally_err,
273                                    } => {
274                                        tracing::warn!(
275                                            finally_error = %finally_err,
276                                            previous_error = %previous,
277                                            "doFinally threw; restoring previous exception (Camel parity)"
278                                        );
279                                        return Err(previous);
280                                    }
281                                };
282                                // AFTER finally has run (and had access to exception props),
283                                // apply handle_error() for Handled disposition to clear the
284                                // error state and set CamelExceptionHandled=true marker.
285                                if matches!(disposition, ExceptionDisposition::Handled) {
286                                    ex.handle_error();
287                                }
288                                match disposition {
289                                    ExceptionDisposition::Handled => Ok(ex),
290                                    _ => Err(original_err),
291                                }
292                            }
293                            Err(failed) => {
294                                // Catch threw. Sealed failure envelope (bd rc-zgbqq):
295                                // the catch error stays the main error (returned
296                                // below and threaded into finally); the original is
297                                // surfaced through the unconditional warn record
298                                // and, best-effort, the active span. The event
299                                // lands on the current span when one is entered —
300                                // no new span, no second event.
301                                let (catch_ex, catch_err) = *failed;
302                                tracing::warn!(
303                                    original_error = %original_err,
304                                    catch_error = %catch_err,
305                                    "do_try catch block failed; catch error supersedes original"
306                                );
307                                record_span_error(&catch_err);
308                                // Run finally with previous=catch_err. Per Camel
309                                // parity, if finally itself throws, catch_err is
310                                // restored.
311                                let outcome = run_finally(
312                                    finally_steps.clone(),
313                                    finally_on_when.clone(),
314                                    catch_ex,
315                                    Some(catch_err.clone()),
316                                )
317                                .await;
318                                if let FinallyOutcome::Restore {
319                                    previous,
320                                    finally_err,
321                                } = outcome
322                                {
323                                    tracing::warn!(
324                                        catch_error = %previous,
325                                        finally_error = %finally_err,
326                                        "doFinally threw after failed catch; restoring catch error"
327                                    );
328                                    return Err(previous);
329                                }
330                                Err(catch_err)
331                            }
332                        };
333                    }
334
335                    // No catch matched. Run finally with previous=original. Propagate original.
336                    let outcome = run_finally(
337                        finally_steps,
338                        finally_on_when,
339                        ex,
340                        Some(original_err.clone()),
341                    )
342                    .await;
343                    if let FinallyOutcome::Restore {
344                        previous,
345                        finally_err,
346                    } = outcome
347                    {
348                        tracing::warn!(
349                            finally_error = %finally_err,
350                            previous_error = %previous,
351                            "doFinally threw; restoring previous exception (Camel parity)"
352                        );
353                        return Err(previous);
354                    }
355                    Err(original_err)
356                }
357            }
358        })
359    }
360}
361
362// ── DoTrySegment (ADR-0025 OutcomePipeline) ──────────────────────────────
363
364/// Compilable segment for a `doCatch` clause within a `DoTrySegment`.
365///
366/// `disposition` controls outcome when the catch body completes:
367#[cfg(test)]
368mod tests {
369    use super::*;
370    use crate::test_log_capture::{capture_debugs_with_span_records, captured_field, record_field};
371    use camel_api::{BoxProcessor, BoxProcessorExt};
372    use std::sync::Arc;
373    use std::sync::atomic::{AtomicU32, Ordering};
374
375    fn passthrough() -> BoxProcessor {
376        BoxProcessor::from_fn(move |ex| Box::pin(async move { Ok(ex) }))
377    }
378
379    fn record_call(flag: Arc<AtomicU32>) -> BoxProcessor {
380        BoxProcessor::from_fn(move |ex| {
381            let f = flag.clone();
382            Box::pin(async move {
383                f.fetch_add(1, Ordering::SeqCst);
384                Ok(ex)
385            })
386        })
387    }
388
389    fn always_fail(err: CamelError) -> BoxProcessor {
390        BoxProcessor::from_fn(move |_ex| {
391            let e = err.clone();
392            Box::pin(async move { Err(e) })
393        })
394    }
395
396    #[tokio::test]
397    async fn happy_path_try_succeeds_finally_runs() {
398        let finally_flag = Arc::new(AtomicU32::new(0));
399        let mut svc = DoTryService::new(vec![passthrough()]);
400        svc.finally_steps = vec![record_call(finally_flag.clone())];
401
402        let mut boxed = BoxProcessor::new(svc);
403        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
404        assert!(result.is_ok());
405        assert_eq!(finally_flag.load(Ordering::SeqCst), 1);
406    }
407
408    #[tokio::test]
409    async fn catch_by_variant_handled_returns_ok() {
410        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
411        let mut svc = DoTryService::new(vec![try_step]);
412        svc.catch_clauses.push(CatchClause {
413            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
414            on_when: None,
415            steps: vec![passthrough()],
416            disposition: ExceptionDisposition::Handled,
417        });
418
419        let mut boxed = BoxProcessor::new(svc);
420        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
421        assert!(result.is_ok(), "Handled must return Ok");
422        let ex = result.unwrap();
423        assert_eq!(
424            ex.properties.get(PROPERTY_EXCEPTION_HANDLED),
425            Some(&camel_api::Value::Bool(true)),
426            "CamelExceptionHandled must be set via handle_error()"
427        );
428    }
429
430    #[tokio::test]
431    async fn catch_by_variant_propagate_runs_side_effects_and_rethrows() {
432        let original = CamelError::ProcessorError("boom".into());
433        let try_step = always_fail(original.clone());
434        let side_effect = Arc::new(AtomicU32::new(0));
435        let catch_step = record_call(side_effect.clone());
436        let mut svc = DoTryService::new(vec![try_step]);
437        svc.catch_clauses.push(CatchClause {
438            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
439            on_when: None,
440            steps: vec![catch_step],
441            disposition: ExceptionDisposition::Propagate,
442        });
443
444        let mut boxed = BoxProcessor::new(svc);
445        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
446        assert!(result.is_err(), "Propagate must rethrow original");
447        assert!(matches!(result.unwrap_err(), CamelError::ProcessorError(_)));
448        assert_eq!(
449            side_effect.load(Ordering::SeqCst),
450            1,
451            "catch branch must have run for side-effects"
452        );
453    }
454
455    #[tokio::test]
456    async fn catch_by_predicate_matches_via_exception_kind() {
457        let try_step = always_fail(CamelError::Io("disk full".into()));
458        let predicate = FilterPredicate::new(|ex: &Exchange| {
459            ex.properties
460                .get(camel_api::exchange::PROPERTY_EXCEPTION_KIND)
461                .map(|v| matches!(v, camel_api::Value::String(s) if s == "io"))
462                .unwrap_or(false)
463        });
464        let mut svc = DoTryService::new(vec![try_step]);
465        svc.catch_clauses.push(CatchClause {
466            matcher: CatchMatcher::Predicate(predicate),
467            on_when: None,
468            steps: vec![passthrough()],
469            disposition: ExceptionDisposition::Handled,
470        });
471
472        let mut boxed = BoxProcessor::new(svc);
473        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
474        assert!(
475            result.is_ok(),
476            "Predicate matcher must catch the error and Handled must return Ok"
477        );
478    }
479
480    #[tokio::test]
481    async fn on_when_filters_clause_and_next_evaluated() {
482        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
483        let first_call = Arc::new(AtomicU32::new(0));
484        let second_call = Arc::new(AtomicU32::new(0));
485
486        let mut svc = DoTryService::new(vec![try_step]);
487        svc.catch_clauses.push(CatchClause {
488            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
489            on_when: Some(FilterPredicate::new(|_ex| false)),
490            steps: vec![record_call(first_call.clone())],
491            disposition: ExceptionDisposition::Handled,
492        });
493        svc.catch_clauses.push(CatchClause {
494            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
495            on_when: None,
496            steps: vec![record_call(second_call.clone())],
497            disposition: ExceptionDisposition::Handled,
498        });
499
500        let mut boxed = BoxProcessor::new(svc);
501        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
502        assert_eq!(first_call.load(Ordering::SeqCst), 0);
503        assert_eq!(second_call.load(Ordering::SeqCst), 1);
504    }
505
506    #[tokio::test]
507    async fn first_match_wins_subsequent_clauses_not_evaluated() {
508        let try_step = always_fail(CamelError::Io("err".into()));
509        let first_call = Arc::new(AtomicU32::new(0));
510        let second_call = Arc::new(AtomicU32::new(0));
511
512        let mut svc = DoTryService::new(vec![try_step]);
513        svc.catch_clauses.push(CatchClause {
514            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
515            on_when: None,
516            steps: vec![record_call(first_call.clone())],
517            disposition: ExceptionDisposition::Handled,
518        });
519        svc.catch_clauses.push(CatchClause {
520            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
521            on_when: None,
522            steps: vec![record_call(second_call.clone())],
523            disposition: ExceptionDisposition::Handled,
524        });
525
526        let mut boxed = BoxProcessor::new(svc);
527        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
528        assert_eq!(first_call.load(Ordering::SeqCst), 1);
529        assert_eq!(second_call.load(Ordering::SeqCst), 0);
530    }
531
532    #[tokio::test]
533    async fn no_clause_matches_propagates_original() {
534        let try_step = always_fail(CamelError::CircuitOpen("cb".into()));
535        let mut svc = DoTryService::new(vec![try_step]);
536        svc.catch_clauses.push(CatchClause {
537            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
538            on_when: None,
539            steps: vec![passthrough()],
540            disposition: ExceptionDisposition::Handled,
541        });
542
543        let mut boxed = BoxProcessor::new(svc);
544        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
545        assert!(result.is_err());
546        assert!(matches!(result.unwrap_err(), CamelError::CircuitOpen(_)));
547    }
548
549    #[tokio::test]
550    async fn catch_branch_throws_new_error_wins() {
551        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
552        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
553        let mut svc = DoTryService::new(vec![try_step]);
554        svc.catch_clauses.push(CatchClause {
555            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
556            on_when: None,
557            steps: vec![catch_step],
558            disposition: ExceptionDisposition::Handled,
559        });
560
561        let mut boxed = BoxProcessor::new(svc);
562        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
563        assert!(result.is_err());
564        assert!(matches!(result.unwrap_err(), CamelError::Io(_)));
565    }
566
567    #[tokio::test]
568    async fn finally_throws_with_no_previous_error_propagates_finally_error() {
569        let finally_step = always_fail(CamelError::Config("fin".into()));
570        let mut svc = DoTryService::new(vec![passthrough()]);
571        svc.finally_steps = vec![finally_step];
572
573        let mut boxed = BoxProcessor::new(svc);
574        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
575        assert!(result.is_err());
576        assert!(matches!(result.unwrap_err(), CamelError::Config(_)));
577    }
578
579    #[tokio::test]
580    async fn finally_throws_with_previous_error_restores_previous() {
581        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
582        let finally_step = always_fail(CamelError::Config("fin".into()));
583        let mut svc = DoTryService::new(vec![try_step]);
584        svc.finally_steps = vec![finally_step];
585
586        let mut boxed = BoxProcessor::new(svc);
587        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
588        assert!(result.is_err());
589        assert!(
590            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
591            "previous error must be restored when finally throws (Camel parity)"
592        );
593    }
594
595    #[tokio::test]
596    async fn finally_on_when_false_skips_finally() {
597        let finally_call = Arc::new(AtomicU32::new(0));
598        let mut svc = DoTryService::new(vec![passthrough()]);
599        svc.finally_steps = vec![record_call(finally_call.clone())];
600        svc.finally_on_when = Some(FilterPredicate::new(|_ex| false));
601
602        let mut boxed = BoxProcessor::new(svc);
603        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
604        assert_eq!(finally_call.load(Ordering::SeqCst), 0);
605    }
606
607    #[tokio::test]
608    async fn stale_handled_marker_cleared_on_entry() {
609        let mut ex = Exchange::default();
610        ex.set_property(PROPERTY_EXCEPTION_HANDLED, camel_api::Value::Bool(true));
611        let svc = DoTryService::new(vec![passthrough()]);
612        let mut boxed = BoxProcessor::new(svc);
613        let result = boxed.ready().await.unwrap().call(ex).await;
614        let ex = result.unwrap();
615        assert!(
616            !ex.properties.contains_key(PROPERTY_EXCEPTION_HANDLED),
617            "stale CamelExceptionHandled must be cleared on entry"
618        );
619    }
620
621    #[tokio::test]
622    async fn nested_do_try_inner_catch_does_not_leak_to_outer() {
623        let inner = {
624            let try_step = always_fail(CamelError::Io("inner".into()));
625            let mut d = DoTryService::new(vec![try_step]);
626            d.catch_clauses.push(CatchClause {
627                matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
628                on_when: None,
629                steps: vec![passthrough()],
630                disposition: ExceptionDisposition::Handled,
631            });
632            BoxProcessor::new(d)
633        };
634        let mut outer = DoTryService::new(vec![inner]);
635        outer.catch_clauses.push(CatchClause {
636            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
637            on_when: None,
638            steps: vec![passthrough()],
639            disposition: ExceptionDisposition::Handled,
640        });
641
642        let mut boxed = BoxProcessor::new(outer);
643        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
644        assert!(
645            result.is_ok(),
646            "outer must see Ok because inner handled its own error"
647        );
648    }
649
650    #[tokio::test]
651    async fn catch_all_only_fires_when_no_specific_clause_matches() {
652        let try_step = always_fail(CamelError::Io("err".into()));
653        let processor_call = Arc::new(AtomicU32::new(0));
654        let catch_all_call = Arc::new(AtomicU32::new(0));
655
656        let mut svc = DoTryService::new(vec![try_step]);
657        // First clause (specific) targets ProcessorError — won't match Io error.
658        svc.catch_clauses.push(CatchClause {
659            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
660            on_when: None,
661            steps: vec![record_call(processor_call.clone())],
662            disposition: ExceptionDisposition::Handled,
663        });
664        // Second clause is the catch-all — should fire.
665        svc.catch_clauses.push(CatchClause {
666            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
667            on_when: None,
668            steps: vec![record_call(catch_all_call.clone())],
669            disposition: ExceptionDisposition::Handled,
670        });
671
672        let mut boxed = BoxProcessor::new(svc);
673        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
674        assert_eq!(
675            processor_call.load(Ordering::SeqCst),
676            0,
677            "specific ProcessorError clause must not fire on Io error"
678        );
679        assert_eq!(
680            catch_all_call.load(Ordering::SeqCst),
681            1,
682            "catch-all clause must fire when no specific clause matches"
683        );
684    }
685
686    #[tokio::test]
687    async fn catch_throws_with_finally_runs_finally_and_propagates_catch_err() {
688        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
689        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
690        let finally_flag = Arc::new(AtomicU32::new(0));
691        let finally_step = record_call(finally_flag.clone());
692
693        let mut svc = DoTryService::new(vec![try_step]);
694        svc.catch_clauses.push(CatchClause {
695            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
696            on_when: None,
697            steps: vec![catch_step],
698            disposition: ExceptionDisposition::Handled,
699        });
700        svc.finally_steps = vec![finally_step];
701
702        let mut boxed = BoxProcessor::new(svc);
703        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
704
705        assert!(result.is_err());
706        assert!(
707            matches!(result.unwrap_err(), CamelError::Io(_)),
708            "catch_err must propagate (not original ProcessorError)"
709        );
710        assert_eq!(
711            finally_flag.load(Ordering::SeqCst),
712            1,
713            "doFinally must run even when catch throws"
714        );
715    }
716
717    #[tokio::test]
718    async fn catch_throws_and_finally_throws_restores_catch_err() {
719        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
720        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
721        let finally_step = always_fail(CamelError::Config("fin-fail".into()));
722
723        let mut svc = DoTryService::new(vec![try_step]);
724        svc.catch_clauses.push(CatchClause {
725            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
726            on_when: None,
727            steps: vec![catch_step],
728            disposition: ExceptionDisposition::Handled,
729        });
730        svc.finally_steps = vec![finally_step];
731
732        let mut boxed = BoxProcessor::new(svc);
733        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
734
735        assert!(result.is_err());
736        assert!(
737            matches!(result.unwrap_err(), CamelError::Io(_)),
738            "catch_err (Io) must be restored over finally_err (Config) per Camel parity"
739        );
740    }
741
742    #[tokio::test]
743    async fn finally_on_when_false_with_previous_error_still_propagates_original() {
744        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
745        let finally_flag = Arc::new(AtomicU32::new(0));
746        let finally_step = record_call(finally_flag.clone());
747
748        let mut svc = DoTryService::new(vec![try_step]);
749        // No catch clauses → original error stays.
750        svc.finally_steps = vec![finally_step];
751        svc.finally_on_when = Some(FilterPredicate::new(|_ex| false));
752
753        let mut boxed = BoxProcessor::new(svc);
754        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
755
756        assert!(result.is_err());
757        assert!(
758            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
759            "original error must propagate even when finally_on_when skips finally"
760        );
761        assert_eq!(
762            finally_flag.load(Ordering::SeqCst),
763            0,
764            "doFinally must NOT run when on_when returns false"
765        );
766    }
767
768    fn catch_fails_service() -> BoxProcessor {
769        let try_step = always_fail(CamelError::ProcessorError("orig-lost".into()));
770        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
771        let mut svc = DoTryService::new(vec![try_step]);
772        svc.catch_clauses.push(CatchClause {
773            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
774            on_when: None,
775            steps: vec![catch_step],
776            disposition: ExceptionDisposition::Handled,
777        });
778        BoxProcessor::new(svc)
779    }
780
781    #[tokio::test]
782    async fn catch_throws_under_propagate_disposition_returns_catch_err() {
783        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
784        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
785        let mut svc = DoTryService::new(vec![try_step]);
786        svc.catch_clauses.push(CatchClause {
787            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
788            on_when: None,
789            steps: vec![catch_step],
790            disposition: ExceptionDisposition::Propagate,
791        });
792
793        let mut boxed = BoxProcessor::new(svc);
794        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
795        assert!(
796            matches!(result, Err(CamelError::Io(_))),
797            "Propagate-disposition catch failure must return the catch error, \
798             same envelope as Handled, got: {result:?}"
799        );
800    }
801
802    #[test]
803    fn catch_throws_logs_original_and_catch_error() {
804        let mut boxed = catch_fails_service();
805        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
806            tokio::runtime::Builder::new_current_thread()
807                .enable_all()
808                .build()
809                .expect("current-thread runtime")
810                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
811        });
812
813        assert!(
814            matches!(result, Err(CamelError::Io(_))),
815            "catch failure must return Err(Io), got: {result:?}"
816        );
817        // Structured field lookup on the envelope record, not message
818        // formatting.
819        let original = record_field(&captured, "do_try catch block failed", "original_error")
820            .unwrap_or_else(|| {
821                panic!("envelope record missing original_error field, captured: {captured:?}")
822            });
823        let catch = record_field(&captured, "do_try catch block failed", "catch_error")
824            .unwrap_or_else(|| {
825                panic!("envelope record missing catch_error field, captured: {captured:?}")
826            });
827        assert!(
828            original.contains("orig-lost"),
829            "original_error must carry the original error, got: {original}"
830        );
831        assert!(
832            catch.contains("catch-fail"),
833            "catch_error must carry the catch error, got: {catch}"
834        );
835    }
836
837    #[test]
838    fn catch_throws_marks_span_error_and_event() {
839        let mut boxed = catch_fails_service();
840        let (result, captured, span_records) = capture_debugs_with_span_records(|| {
841            // Declared `error` field so record_span_error's record lands.
842            let span = tracing::info_span!("dotry_test", error = tracing::field::Empty);
843            let _guard = span.enter();
844            tokio::runtime::Builder::new_current_thread()
845                .enable_all()
846                .build()
847                .expect("current-thread runtime")
848                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
849        });
850
851        assert!(
852            matches!(result, Err(CamelError::Io(_))),
853            "catch failure must return Err(Io), got: {result:?}"
854        );
855        // The span recorded the `error` field with the CATCH error.
856        // Structured field lookup: substring `error=` would also match
857        // `original_error=` suffixes.
858        assert!(
859            span_records.iter().any(|line| {
860                captured_field(line, "error").is_some_and(|v| v.contains("catch-fail"))
861            }),
862            "expected span error record carrying the catch error, span records: {span_records:?}"
863        );
864        // The WARN event still carries original_error while a span is active.
865        let original = record_field(
866            &captured,
867            "do_try catch block failed",
868            "original_error",
869        )
870        .unwrap_or_else(|| {
871            panic!(
872                "envelope record missing original_error field under active span, captured: {captured:?}"
873            )
874        });
875        assert!(
876            original.contains("orig-lost"),
877            "original_error must carry the original error under an active span, got: {original}"
878        );
879    }
880
881    #[test]
882    fn catch_and_finally_throw_logs_finally_error() {
883        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
884        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
885        let finally_step = always_fail(CamelError::Config("fin-fail".into()));
886        let mut svc = DoTryService::new(vec![try_step]);
887        svc.catch_clauses.push(CatchClause {
888            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
889            on_when: None,
890            steps: vec![catch_step],
891            disposition: ExceptionDisposition::Handled,
892        });
893        svc.finally_steps = vec![finally_step];
894
895        let mut boxed = BoxProcessor::new(svc);
896        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
897            tokio::runtime::Builder::new_current_thread()
898                .enable_all()
899                .build()
900                .expect("current-thread runtime")
901                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
902        });
903
904        assert!(
905            matches!(result, Err(CamelError::Io(_))),
906            "catch error must be restored over finally error, got: {result:?}"
907        );
908        let catch = record_field(
909            &captured,
910            "doFinally threw after failed catch; restoring catch error",
911            "catch_error",
912        )
913        .unwrap_or_else(|| {
914            panic!("catch-failed restore record missing catch_error field, captured: {captured:?}")
915        });
916        let finally = record_field(
917            &captured,
918            "doFinally threw after failed catch; restoring catch error",
919            "finally_error",
920        )
921        .unwrap_or_else(|| {
922            panic!(
923                "catch-failed restore record missing finally_error field, captured: {captured:?}"
924            )
925        });
926        assert!(
927            catch.contains("catch-fail"),
928            "catch_error must carry the catch error, got: {catch}"
929        );
930        assert!(
931            finally.contains("fin-fail"),
932            "finally_error must carry the finally error, got: {finally}"
933        );
934    }
935}
camel_processor/do_try.rs

camel_processor/
do_try.rs
