Skip to main content

camel_processor/
do_try_segment.rs

1//! ## Stop semantics (ADR-0025)
2//!
3//! This segment implements `OutcomePipeline` and propagates `PipelineOutcome::Stopped(ex)`
4//! with the exchange state intact. See ADR-0025 §3.
5
6use crate::do_try::CatchMatcher;
7use crate::do_try::chain_predicate_error;
8use crate::error_handler::record_span_error;
9use camel_api::error_handler::ExceptionDisposition;
10use camel_api::outcome_pipeline::OutcomePipeline;
11use camel_api::pipeline_outcome::PipelineOutcome;
12use camel_api::{CamelError, Exchange, PredicateSource};
13use std::future::Future;
14use std::pin::Pin;
15
16/// - `Handled`: the exchange continues through finally and downstream.
17/// - `Propagate`: the catch body runs for side effects, finally runs with the
18///   catch body's exchange, and the original try-error re-throws as `Failed`.
19#[derive(Clone)]
20pub struct CatchClauseSegment {
21    pub matcher: CatchMatcher,
22    pub on_when: Option<PredicateSource>,
23    pub body: camel_api::OutcomeSegment,
24    pub disposition: ExceptionDisposition,
25}
26
27/// Compilable segment for a `doFinally` clause within a `DoTrySegment`.
28#[derive(Clone)]
29pub struct FinallyClauseSegment {
30    pub on_when: Option<PredicateSource>,
31    pub body: camel_api::OutcomeSegment,
32}
33
34/// Outcome-aware structural EIP segment for the doTry/doCatch/doFinally pattern.
35///
36/// Operates at the `PipelineOutcome` layer so that `Stopped(ex)` from a
37/// sub-step (e.g. Stop EIP) is preserved with the exchange including all
38/// mutations. See ADR-0025 §5.1 for full semantics.
39pub struct DoTrySegment {
40    pub try_body: camel_api::OutcomeSegment,
41    pub catches: Vec<CatchClauseSegment>,
42    pub finally: Option<FinallyClauseSegment>,
43}
44
45impl Clone for DoTrySegment {
46    fn clone(&self) -> Self {
47        Self {
48            try_body: self.try_body.clone(),
49            catches: self.catches.clone(),
50            finally: self.finally.clone(),
51        }
52    }
53}
54
55/// Narrow outcome type for `run_finally_body` so the compiler enforces
56/// exhaustiveness at the two call sites instead of `unreachable!()`.
57/// This is a private, transient type; the `Stopped` exchange is boxed to keep
58/// the `Err` variant of `run_finally_body` small (clippy::result_large_err).
59enum FinallyOutcome {
60    Stopped(Box<Exchange>),
61    Failed(CamelError),
62    /// The finally `on_when` predicate itself failed; the typed error fails
63    /// the segment.
64    PredicateFailed(CamelError),
65}
66
67/// Run the finally body if present and on_when permits.
68/// Free function (not a method) to avoid borrow conflict with
69/// `self.catches.iter_mut()` inside the catch loop.
70/// The `FinallyOutcome` variants move the `Exchange` (the catch loop needs
71/// its state); boxing would rewrite the segment call chain for a lint
72/// threshold.
73#[allow(clippy::result_large_err)]
74async fn run_finally_body(
75    finally: &mut Option<FinallyClauseSegment>,
76    ex: Exchange,
77) -> Result<Exchange, FinallyOutcome> {
78    let Some(f) = finally.as_mut() else {
79        return Ok(ex);
80    };
81    if let Some(on_when) = &f.on_when {
82        match on_when.matches(&ex).await {
83            Ok(true) => {}
84            Ok(false) => return Ok(ex),
85            Err(err) => return Err(FinallyOutcome::PredicateFailed(err)),
86        }
87    }
88    match f.body.run(ex).await {
89        PipelineOutcome::Completed(e) => Ok(e),
90        PipelineOutcome::Stopped(e) => Err(FinallyOutcome::Stopped(Box::new(e))),
91        PipelineOutcome::Failed(e) => Err(FinallyOutcome::Failed(e)),
92    }
93}
94
95/// Surface `failure` after running finally.
96///
97/// Used for a failed catch predicate (the predicate error surfaced as the
98/// main failure), the no-catch-matched path, and the Propagate / fail-closed
99/// path (the original try error).
100/// `failure` becomes the main failure (Camel parity: a throwing finally body
101/// restores it; Stop in finally still stops the outer route; a failed finally
102/// `on_when` predicate supersedes it).
103async fn fail_after_finally(
104    finally: &mut Option<FinallyClauseSegment>,
105    ex: Exchange,
106    failure: CamelError,
107) -> PipelineOutcome {
108    match run_finally_body(finally, ex).await {
109        Ok(_) => PipelineOutcome::Failed(failure),
110        Err(FinallyOutcome::Stopped(e)) => PipelineOutcome::Stopped(*e),
111        Err(FinallyOutcome::Failed(_finally_err)) => {
112            tracing::warn!(
113                error = %failure,
114                "doFinally threw; restoring previous error (Camel parity)"
115            );
116            PipelineOutcome::Failed(failure)
117        }
118        Err(FinallyOutcome::PredicateFailed(pred_err)) => PipelineOutcome::Failed(pred_err),
119    }
120}
121
122impl OutcomePipeline for DoTrySegment {
123    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
124        Box::new(self.clone())
125    }
126
127    fn run<'a>(
128        &'a mut self,
129        exchange: Exchange,
130    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
131        Box::pin(async move {
132            // Pre-clone for the "no catch matched" path; `exchange` itself is
133            // consumed by `try_body.run(...)` below.
134            let exchange_for_unmatched = exchange.branch_clone();
135
136            // 1. Run try_body. If Stopped, return immediately (skip catch AND finally).
137            let try_outcome = self.try_body.run(exchange).await;
138            let returned_ex = match try_outcome {
139                PipelineOutcome::Stopped(ex) => return PipelineOutcome::Stopped(ex),
140                PipelineOutcome::Completed(ex) => ex,
141                PipelineOutcome::Failed(err) => {
142                    // 2. try failed — try each catch in order. Walk the catch
143                    // chain starting from the failed try-exchange. On the first
144                    // matching catch whose `on_when` is true (or unset), run it.
145                    // A failed catch `when`/`on_when` predicate replaces the
146                    // original error (preserved as `cause`) and fails the scope.
147                    let mut current_ex = exchange_for_unmatched;
148                    current_ex.set_error(err.clone());
149                    // Distinguishes "a catch handled the error" (loop breaks
150                    // with `current_ex` = catch output) from "the loop
151                    // exhausted with no match" (must propagate the original).
152                    let mut catch_matched = false;
153                    for catch in self.catches.iter_mut() {
154                        let matched = match catch.matcher.matches(&err, &current_ex).await {
155                            Ok(m) => m,
156                            Err(pred_err) => {
157                                let chained = chain_predicate_error(pred_err, err);
158                                return fail_after_finally(&mut self.finally, current_ex, chained)
159                                    .await;
160                            }
161                        };
162                        if !matched {
163                            continue;
164                        }
165                        if let Some(on_when) = &catch.on_when {
166                            match on_when.matches(&current_ex).await {
167                                Ok(true) => {}
168                                Ok(false) => continue,
169                                Err(pred_err) => {
170                                    let chained = chain_predicate_error(pred_err, err);
171                                    return fail_after_finally(
172                                        &mut self.finally,
173                                        current_ex,
174                                        chained,
175                                    )
176                                    .await;
177                                }
178                            }
179                        }
180                        match catch.body.run(current_ex).await {
181                            // Stop in catch: skip finally AND outer route halts.
182                            PipelineOutcome::Stopped(stopped_ex) => {
183                                return PipelineOutcome::Stopped(stopped_ex);
184                            }
185                            // Catch handled — exit catch chain with this exchange.
186                            PipelineOutcome::Completed(next) => {
187                                match catch.disposition {
188                                    ExceptionDisposition::Handled => {
189                                        current_ex = next;
190                                        catch_matched = true;
191                                        break;
192                                    }
193                                    ExceptionDisposition::Continued => {
194                                        current_ex = next;
195                                        catch_matched = true;
196                                        break;
197                                    }
198                                    // Propagate and any future variant run finally then
199                                    // surface the original error (fail-closed).
200                                    _ => {
201                                        return fail_after_finally(&mut self.finally, next, err)
202                                            .await;
203                                    }
204                                }
205                            }
206                            // Catch-body Failed: surface THAT error to outer
207                            // route. Per ADR-0025 invariant #4 ("doTry is a
208                            // local error-handler island"), a failing catch
209                            // body propagates as Failed — it does NOT re-enter
210                            // the catch chain (no recursive catch-of-catch in
211                            // Camel). Skip remaining catches and finally.
212                            PipelineOutcome::Failed(catch_err) => {
213                                // Sealed failure envelope: the catch error
214                                // stays the main error (returned below);
215                                // the original try error is surfaced
216                                // through the unconditional warn record
217                                // and, when a span is active, the span
218                                // error record. No new span is created.
219                                tracing::warn!(
220                                    original_error = %err,
221                                    catch_error = %catch_err,
222                                    "do_try catch block failed; catch error supersedes original"
223                                );
224                                record_span_error(&catch_err);
225                                return PipelineOutcome::Failed(catch_err);
226                            }
227                        }
228                    }
229                    if catch_matched {
230                        // A catch handled the error: continue to the shared
231                        // finally path and return its exchange as Completed.
232                        current_ex
233                    } else {
234                        // No catch matched: run finally, then propagate the
235                        // original try error as `Failed`. Falling through to the
236                        // shared finally path would return `Completed` and
237                        // swallow the error, letting the route continue past
238                        // the scope (mirrors `DoTryService`'s no-match arm).
239                        return fail_after_finally(&mut self.finally, current_ex, err).await;
240                    }
241                }
242            };
243            // 3. Run finally if present (skip if try/catch returned Stopped —
244            // already returned above; skip if catch-body Failed — also returned).
245            match run_finally_body(&mut self.finally, returned_ex).await {
246                Ok(ex) => PipelineOutcome::Completed(ex),
247                Err(FinallyOutcome::Stopped(e)) => PipelineOutcome::Stopped(*e),
248                Err(FinallyOutcome::Failed(finally_err)) => {
249                    tracing::warn!(
250                        error = %finally_err,
251                        "doFinally threw during/after catch; surfacing finally error"
252                    );
253                    PipelineOutcome::Failed(finally_err)
254                }
255                // Failed finally `on_when` predicate: the typed error fails
256                // the segment.
257                Err(FinallyOutcome::PredicateFailed(pred_err)) => PipelineOutcome::Failed(pred_err),
258            }
259        })
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use super::*;
266    use crate::do_try::CatchMatcher;
267    use crate::test_log_capture::{capture_debugs_with_span_records, captured_field, record_field};
268    use camel_api::pipeline_outcome::PipelineOutcome;
269    use std::sync::Arc;
270    use std::sync::atomic::{AtomicU32, Ordering};
271
272    // ── Helpers for DoTrySegment tests ──
273
274    struct CompleteSegment;
275
276    impl OutcomePipeline for CompleteSegment {
277        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
278            Box::new(CompleteSegment)
279        }
280        fn run<'a>(
281            &'a mut self,
282            exchange: Exchange,
283        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
284            Box::pin(async move { PipelineOutcome::Completed(exchange) })
285        }
286    }
287
288    fn seg_complete() -> camel_api::OutcomeSegment {
289        camel_api::OutcomeSegment::new(Box::new(CompleteSegment))
290    }
291
292    struct FailSegment(CamelError);
293
294    impl OutcomePipeline for FailSegment {
295        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
296            Box::new(FailSegment(self.0.clone()))
297        }
298        fn run<'a>(
299            &'a mut self,
300            _exchange: Exchange,
301        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
302            let e = self.0.clone();
303            Box::pin(async move { PipelineOutcome::Failed(e) })
304        }
305    }
306
307    fn seg_fail(err: CamelError) -> camel_api::OutcomeSegment {
308        camel_api::OutcomeSegment::new(Box::new(FailSegment(err)))
309    }
310
311    struct MutateThenStop {
312        mutator: Arc<dyn Fn(&mut Exchange) + Send + Sync>,
313    }
314
315    impl OutcomePipeline for MutateThenStop {
316        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
317            Box::new(MutateThenStop {
318                mutator: Arc::clone(&self.mutator),
319            })
320        }
321        fn run<'a>(
322            &'a mut self,
323            mut exchange: Exchange,
324        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
325            let m = Arc::clone(&self.mutator);
326            Box::pin(async move {
327                m(&mut exchange);
328                PipelineOutcome::Stopped(exchange)
329            })
330        }
331    }
332
333    fn seg_stop_with(
334        mutator: impl Fn(&mut Exchange) + Send + Sync + 'static,
335    ) -> camel_api::OutcomeSegment {
336        camel_api::OutcomeSegment::new(Box::new(MutateThenStop {
337            mutator: Arc::new(mutator),
338        }))
339    }
340
341    struct RecordCall {
342        counter: Arc<AtomicU32>,
343    }
344
345    impl OutcomePipeline for RecordCall {
346        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
347            Box::new(RecordCall {
348                counter: Arc::clone(&self.counter),
349            })
350        }
351        fn run<'a>(
352            &'a mut self,
353            exchange: Exchange,
354        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
355            let c = Arc::clone(&self.counter);
356            Box::pin(async move {
357                c.fetch_add(1, Ordering::SeqCst);
358                PipelineOutcome::Completed(exchange)
359            })
360        }
361    }
362
363    fn seg_record(counter: Arc<AtomicU32>) -> camel_api::OutcomeSegment {
364        camel_api::OutcomeSegment::new(Box::new(RecordCall { counter }))
365    }
366
367    // ── 5 New tests for DoTrySegment (ADR-0025) ──
368
369    #[tokio::test]
370    async fn stop_inside_try_skips_catch_and_finally() {
371        let catch_call = Arc::new(AtomicU32::new(0));
372        let finally_call = Arc::new(AtomicU32::new(0));
373
374        let mut seg = DoTrySegment {
375            try_body: seg_stop_with(|ex| {
376                ex.set_property("mutated", camel_api::Value::Bool(true));
377            }),
378            catches: vec![CatchClauseSegment {
379                matcher: CatchMatcher::ByVariant(vec!["*".into()]),
380                on_when: None,
381                body: seg_record(catch_call.clone()),
382                disposition: ExceptionDisposition::Handled,
383            }],
384            finally: Some(FinallyClauseSegment {
385                on_when: None,
386                body: seg_record(finally_call.clone()),
387            }),
388        };
389
390        let result = seg.run(Exchange::default()).await;
391        match result {
392            PipelineOutcome::Stopped(ex) => {
393                assert_eq!(
394                    ex.properties.get("mutated"),
395                    Some(&camel_api::Value::Bool(true)),
396                    "try body mutation must be preserved in Stopped exchange"
397                );
398            }
399            other => panic!("expected Stopped, got {:?}", other),
400        }
401        assert_eq!(
402            catch_call.load(Ordering::SeqCst),
403            0,
404            "catch must NOT run when try stops"
405        );
406        assert_eq!(
407            finally_call.load(Ordering::SeqCst),
408            0,
409            "finally must NOT run when try stops"
410        );
411    }
412
413    #[tokio::test]
414    async fn stop_inside_catch_skips_finally() {
415        let finally_call = Arc::new(AtomicU32::new(0));
416
417        let mut seg = DoTrySegment {
418            try_body: seg_fail(CamelError::ProcessorError("boom".into())),
419            catches: vec![CatchClauseSegment {
420                matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
421                on_when: None,
422                body: seg_stop_with(|ex| {
423                    ex.set_property("catch_mutated", camel_api::Value::Bool(true));
424                }),
425                disposition: ExceptionDisposition::Handled,
426            }],
427            finally: Some(FinallyClauseSegment {
428                on_when: None,
429                body: seg_record(finally_call.clone()),
430            }),
431        };
432
433        let result = seg.run(Exchange::default()).await;
434        match result {
435            PipelineOutcome::Stopped(ex) => {
436                assert_eq!(
437                    ex.properties.get("catch_mutated"),
438                    Some(&camel_api::Value::Bool(true)),
439                    "catch body mutation must be preserved in Stopped exchange"
440                );
441            }
442            other => panic!("expected Stopped, got {:?}", other),
443        }
444        assert_eq!(
445            finally_call.load(Ordering::SeqCst),
446            0,
447            "finally must NOT run when catch stops"
448        );
449    }
450
451    #[tokio::test]
452    async fn stop_inside_finally_stops_outer_route() {
453        let mut seg = DoTrySegment {
454            try_body: seg_complete(),
455            catches: vec![],
456            finally: Some(FinallyClauseSegment {
457                on_when: None,
458                body: seg_stop_with(|ex| {
459                    ex.set_property("finally_mutated", camel_api::Value::Bool(true));
460                }),
461            }),
462        };
463
464        let result = seg.run(Exchange::default()).await;
465        match result {
466            PipelineOutcome::Stopped(ex) => {
467                assert_eq!(
468                    ex.properties.get("finally_mutated"),
469                    Some(&camel_api::Value::Bool(true)),
470                    "finally body mutation must be preserved in Stopped exchange"
471                );
472            }
473            other => panic!("expected Stopped, got {:?}", other),
474        }
475    }
476
477    #[tokio::test]
478    async fn catch_on_when_false_falls_through_to_next_catch() {
479        let first_call = Arc::new(AtomicU32::new(0));
480        let second_call = Arc::new(AtomicU32::new(0));
481
482        let mut seg = DoTrySegment {
483            try_body: seg_fail(CamelError::Io("disk err".into())),
484            catches: vec![
485                CatchClauseSegment {
486                    matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
487                    on_when: Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false))),
488                    body: seg_record(first_call.clone()),
489                    disposition: ExceptionDisposition::Handled,
490                },
491                CatchClauseSegment {
492                    matcher: CatchMatcher::ByVariant(vec!["*".into()]),
493                    on_when: None,
494                    body: seg_record(second_call.clone()),
495                    disposition: ExceptionDisposition::Handled,
496                },
497            ],
498            finally: None,
499        };
500
501        let result = seg.run(Exchange::default()).await;
502        assert!(
503            matches!(result, PipelineOutcome::Completed(_)),
504            "expected Completed after second catch"
505        );
506        assert_eq!(
507            first_call.load(Ordering::SeqCst),
508            0,
509            "first catch must NOT fire (on_when=false)"
510        );
511        assert_eq!(
512            second_call.load(Ordering::SeqCst),
513            1,
514            "second catch must fire"
515        );
516    }
517
518    #[tokio::test]
519    async fn finally_on_when_false_skips_finally_entirely() {
520        let finally_call = Arc::new(AtomicU32::new(0));
521        let mut ex = Exchange::default();
522        ex.set_property("try_set", camel_api::Value::Bool(true));
523
524        let mut seg = DoTrySegment {
525            try_body: seg_complete(),
526            catches: vec![],
527            finally: Some(FinallyClauseSegment {
528                on_when: Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false))),
529                body: seg_record(finally_call.clone()),
530            }),
531        };
532
533        let result = seg.run(ex).await;
534        match result {
535            PipelineOutcome::Completed(ex) => {
536                assert_eq!(
537                    ex.properties.get("try_set"),
538                    Some(&camel_api::Value::Bool(true)),
539                    "exchange state from try must be preserved"
540                );
541            }
542            other => panic!("expected Completed, got {:?}", other),
543        }
544        assert_eq!(
545            finally_call.load(Ordering::SeqCst),
546            0,
547            "finally must NOT run when on_when=false"
548        );
549    }
550
551    // ── Catch-body failure envelope (compiled-segment path) ──
552
553    fn catch_fails_segment() -> DoTrySegment {
554        DoTrySegment {
555            try_body: seg_fail(CamelError::ProcessorError("orig".into())),
556            catches: vec![CatchClauseSegment {
557                matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
558                on_when: None,
559                body: seg_fail(CamelError::Io("catch-fail".into())),
560                disposition: ExceptionDisposition::Handled,
561            }],
562            finally: None,
563        }
564    }
565
566    #[test]
567    fn catch_body_failure_returns_catch_err_and_marks_span() {
568        let mut seg = catch_fails_segment();
569        let (result, _captured, span_records) = capture_debugs_with_span_records(|| {
570            // Declared `error` field so record_span_error's record lands.
571            let span = tracing::info_span!("dotry_seg_test", error = tracing::field::Empty);
572            let _guard = span.enter();
573            tokio::runtime::Builder::new_current_thread()
574                .enable_all()
575                .build()
576                .expect("current-thread runtime")
577                .block_on(async { seg.run(Exchange::default()).await })
578        });
579
580        assert!(
581            matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
582            "catch failure must surface the catch error as Failed(Io), got: {result:?}"
583        );
584        // The span recorded the `error` field with the CATCH error.
585        // Structured field lookup: substring `error=` would also match
586        // `original_error=` suffixes.
587        assert!(
588            span_records.iter().any(|line| {
589                captured_field(line, "error").is_some_and(|v| v.contains("catch-fail"))
590            }),
591            "expected span error record carrying the catch error, span records: {span_records:?}"
592        );
593    }
594
595    #[test]
596    fn catch_body_failure_emits_envelope_log() {
597        let mut seg = catch_fails_segment();
598        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
599            tokio::runtime::Builder::new_current_thread()
600                .enable_all()
601                .build()
602                .expect("current-thread runtime")
603                .block_on(async { seg.run(Exchange::default()).await })
604        });
605
606        assert!(
607            matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
608            "catch failure must surface the catch error as Failed(Io), got: {result:?}"
609        );
610        // Structured field lookup on the envelope record, not message
611        // formatting.
612        let original = record_field(&captured, "do_try catch block failed", "original_error")
613            .unwrap_or_else(|| {
614                panic!("envelope record missing original_error field, captured: {captured:?}")
615            });
616        let catch = record_field(&captured, "do_try catch block failed", "catch_error")
617            .unwrap_or_else(|| {
618                panic!("envelope record missing catch_error field, captured: {captured:?}")
619            });
620        assert!(
621            original.contains("orig"),
622            "original_error must carry the original error, got: {original}"
623        );
624        assert!(
625            catch.contains("catch-fail"),
626            "catch_error must carry the catch error, got: {catch}"
627        );
628    }
629
630    // ── Fallible predicate path (language-value-boundary task 1.4) ──
631
632    use camel_api::{ExpressionErrorClass, FilterPredicate, PredicateSource};
633
634    fn expression_failed() -> CamelError {
635        CamelError::ExpressionFailed {
636            language: "rhai".to_string(),
637            route_id: "r1".to_string(),
638            step_id: "step#0".to_string(),
639            verb: "on_when".to_string(),
640            class: ExpressionErrorClass::Runtime,
641            position: None,
642            conversion: None,
643            cause: None,
644        }
645    }
646
647    fn async_err_predicate(err: CamelError) -> PredicateSource {
648        PredicateSource::Async(Arc::new(move |_: &Exchange| {
649            let err = err.clone();
650            Box::pin(async move { Err(err) }) as camel_api::BoxBoolFuture
651        }))
652    }
653
654    #[tokio::test]
655    async fn finally_on_when_predicate_error_fails_segment() {
656        let finally_call = Arc::new(AtomicU32::new(0));
657        let mut seg = DoTrySegment {
658            try_body: seg_complete(),
659            catches: vec![],
660            finally: Some(FinallyClauseSegment {
661                on_when: Some(async_err_predicate(expression_failed())),
662                body: seg_record(finally_call.clone()),
663            }),
664        };
665
666        let result = seg.run(Exchange::default()).await;
667        match result {
668            PipelineOutcome::Failed(CamelError::ExpressionFailed { .. }) => {}
669            other => panic!("expected Failed(ExpressionFailed), got {other:?}"),
670        }
671        assert_eq!(
672            finally_call.load(Ordering::SeqCst),
673            0,
674            "finally body must not run when its on_when predicate errors"
675        );
676    }
677
678    // ── No-catch-matched path (mission 353) ──
679
680    /// doTry with no matching catch must run finally and propagate the
681    /// original try error as `Failed` — the segment counterpart of
682    /// `DoTryService::no_clause_matches_propagates_original`. Guards against
683    /// regressing back to `Completed(bugged_swallow)`.
684    #[tokio::test]
685    async fn no_clause_matches_runs_finally_and_propagates_original() {
686        let catch_call = Arc::new(AtomicU32::new(0));
687        let finally_call = Arc::new(AtomicU32::new(0));
688
689        let mut seg = DoTrySegment {
690            try_body: seg_fail(CamelError::EndpointCreationFailed(
691                "direct endpoint 'direct:does-not-exist' not registered".into(),
692            )),
693            catches: vec![CatchClauseSegment {
694                matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
695                on_when: None,
696                body: seg_record(catch_call.clone()),
697                disposition: ExceptionDisposition::Handled,
698            }],
699            finally: Some(FinallyClauseSegment {
700                on_when: None,
701                body: seg_record(finally_call.clone()),
702            }),
703        };
704
705        let result = seg.run(Exchange::default()).await;
706        match result {
707            PipelineOutcome::Failed(CamelError::EndpointCreationFailed(msg)) => {
708                assert!(
709                    msg.contains("does-not-exist"),
710                    "propagated error must be the original try error, got: {msg}"
711                );
712            }
713            other => {
714                panic!("no catch matched: expected Failed(EndpointCreationFailed), got {other:?}")
715            }
716        }
717        assert_eq!(
718            finally_call.load(Ordering::SeqCst),
719            1,
720            "finally must run exactly once when no catch matches"
721        );
722        assert_eq!(
723            catch_call.load(Ordering::SeqCst),
724            0,
725            "the unmatched catch body must not run"
726        );
727    }
728}