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, PredicateSource};
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 a fallible predicate over the Exchange.
23    Predicate(PredicateSource),
24}
25
26impl CatchMatcher {
27    /// Returns whether this matcher matches the given error and exchange.
28    ///
29    /// A failed predicate evaluation yields `Err` (language-value-boundary:
30    /// the caller chains the original error and fails the scope).
31    pub async fn matches(&self, err: &CamelError, ex: &Exchange) -> Result<bool, CamelError> {
32        match self {
33            CatchMatcher::ByVariant(names) => {
34                if names.iter().any(|n| n == "*") {
35                    return Ok(true);
36                }
37                Ok(names.iter().any(|n| n == err.variant_name()))
38            }
39            CatchMatcher::Predicate(source) => source.matches(ex).await,
40        }
41    }
42}
43
44/// Chain the original try error into a failed catch-predicate error
45/// (language-value-boundary: catch `when` / `on_when` failure envelope).
46///
47/// The copy keeps every field unchanged except `cause`, which holds the
48/// original error so the chain stays retrievable. Two kinds of cause are
49/// replaced: an empty slot (`cause: None`) and a mapper-generated diagnostic
50/// cause identified by the camel-language-api provenance marker (never by
51/// variant or string shape). Genuine caller-supplied chains — including
52/// hand-built `ProcessorError`/`ProcessorErrorWithSource` causes — pass
53/// through untouched. Errors that carry no `cause` slot pass through
54/// untouched as well.
55pub(crate) fn chain_predicate_error(
56    mut predicate_err: CamelError,
57    original: CamelError,
58) -> CamelError {
59    let should_chain = match &predicate_err {
60        CamelError::ExpressionFailed { cause, .. } => {
61            cause.is_none() || camel_language_api::is_expression_detail_cause(&predicate_err)
62        }
63        _ => false,
64    };
65    if should_chain && let CamelError::ExpressionFailed { cause, .. } = &mut predicate_err {
66        *cause = Some(Box::new(original));
67    }
68    predicate_err
69}
70
71/// A single `doCatch` clause.
72#[derive(Clone)]
73pub struct CatchClause {
74    /// The main matcher (variant-name list or predicate).
75    pub matcher: CatchMatcher,
76    /// Optional sub-predicate evaluated AFTER the main matcher passes.
77    pub on_when: Option<PredicateSource>,
78    /// Sub-pipeline executed when the clause matches.
79    pub steps: Vec<BoxProcessor>,
80    /// ADR-0019 disposition: Handled (default), Propagate, or Continued (rejected at parse time).
81    /// In YAML, use lowercase: `handled`, `propagate`, `continued`.
82    pub disposition: ExceptionDisposition,
83}
84
85/// The `doTry` processor. Wrap with `BoxProcessor::new(DoTryService::new(...))`.
86#[derive(Clone)]
87pub struct DoTryService {
88    /// Steps in the try block.
89    pub try_steps: Vec<BoxProcessor>,
90    /// Catch clauses evaluated first-match-wins.
91    pub catch_clauses: Vec<CatchClause>,
92    /// Steps in the finally block (empty = no finally).
93    pub finally_steps: Vec<BoxProcessor>,
94    /// Optional onWhen predicate for finally.
95    pub finally_on_when: Option<PredicateSource>,
96}
97
98impl DoTryService {
99    /// Create a new `DoTryService` with the given try steps.
100    pub fn new(try_steps: Vec<BoxProcessor>) -> Self {
101        Self {
102            try_steps,
103            catch_clauses: Vec::new(),
104            finally_steps: Vec::new(),
105            finally_on_when: None,
106        }
107    }
108
109    /// Full constructor used by the compile pipeline (Task 10b control_flow.rs).
110    /// Builder API (Task 8) constructs via `new()` + field mutation.
111    pub fn with_catch_and_finally(
112        try_steps: Vec<BoxProcessor>,
113        catch_clauses: Vec<CatchClause>,
114        finally_steps: Vec<BoxProcessor>,
115        finally_on_when: Option<PredicateSource>,
116    ) -> Self {
117        Self {
118            try_steps,
119            catch_clauses,
120            finally_steps,
121            finally_on_when,
122        }
123    }
124}
125
126/// Run a sequence of steps, preserving the last exchange state on error.
127/// Returns `Err(Box<(last_ex, err)>)` so DoTry can populate exception properties.
128async fn run_pipeline(
129    steps: Vec<BoxProcessor>,
130    mut ex: Exchange,
131) -> Result<Exchange, Box<(Exchange, CamelError)>> {
132    for mut svc in steps {
133        match svc.ready().await {
134            Ok(ready) => {
135                let snapshot = ex.branch_clone();
136                match ready.call(ex).await {
137                    Ok(new_ex) => ex = new_ex,
138                    Err(err) => return Err(Box::new((snapshot, err))),
139                }
140            }
141            Err(err) => return Err(Box::new((ex, err))),
142        }
143    }
144    Ok(ex)
145}
146
147/// Run the finally block. Camel parity for finally-throws:
148/// - If finally succeeds (or is skipped): `Completed` with its exchange.
149/// - If finally throws AND there was a previous error: `Restore` (caller
150///   logs and restores the previous error).
151/// - If finally throws AND no previous error: `NoPreviousFail` (caller logs
152///   and propagates finally_err).
153/// - If the finally `on_when` predicate itself fails: `Err` (the typed
154///   predicate error fails the scope).
155///
156/// Logging lives at the CALLERS so the restore record's field names can
157/// match the calling flow (`previous_error` vs `catch_error`, bd rc-zgbqq).
158async fn run_finally(
159    finally_steps: Vec<BoxProcessor>,
160    finally_on_when: Option<PredicateSource>,
161    ex: Exchange,
162    previous_err: Option<CamelError>,
163) -> Result<FinallyOutcome, CamelError> {
164    if finally_steps.is_empty() {
165        return Ok(FinallyOutcome::Completed(ex));
166    }
167    if let Some(on_when) = &finally_on_when
168        && !on_when.matches(&ex).await?
169    {
170        return Ok(FinallyOutcome::Completed(ex));
171    }
172    match run_pipeline(finally_steps, ex).await {
173        Ok(ex) => Ok(FinallyOutcome::Completed(ex)),
174        Err(failed) => {
175            let (_, finally_err) = *failed;
176            Ok(match previous_err {
177                Some(prev) => FinallyOutcome::Restore {
178                    previous: prev,
179                    finally_err,
180                },
181                None => FinallyOutcome::NoPreviousFail(finally_err),
182            })
183        }
184    }
185}
186
187/// Result of `run_finally`. Tower-local to this module — unrelated to the
188/// identically-named enum in `do_try_segment.rs`.
189// The variant shape is fixed by the sealed do_try ruling (bd rc-zgbqq);
190// Exchange simply dwarfs the error payloads, so allow the lint instead of
191// boxing and drifting from the specified envelope.
192#[allow(clippy::large_enum_variant)]
193enum FinallyOutcome {
194    /// Finally ran (or was skipped) successfully; carries its exchange.
195    Completed(Exchange),
196    /// Finally threw with no previous error; the finally error wins.
197    NoPreviousFail(CamelError),
198    /// Finally threw with a previous error; the previous error is restored.
199    Restore {
200        previous: CamelError,
201        finally_err: CamelError,
202    },
203}
204
205/// Run finally after a catch-predicate failure, then surface `failure`.
206///
207/// A failed finally `on_when` predicate supersedes `failure`; a throwing
208/// finally body restores `failure` (Camel parity with the other restore
209/// flows).
210async fn run_finally_for_failure(
211    finally_steps: Vec<BoxProcessor>,
212    finally_on_when: Option<PredicateSource>,
213    ex: Exchange,
214    failure: CamelError,
215) -> CamelError {
216    match run_finally(finally_steps, finally_on_when, ex, Some(failure.clone())).await {
217        Err(pred_err) => pred_err,
218        Ok(FinallyOutcome::Restore {
219            previous,
220            finally_err,
221        }) => {
222            tracing::warn!(
223                finally_error = %finally_err,
224                previous_error = %previous,
225                "doFinally threw after catch predicate failure; restoring previous (Camel parity)"
226            );
227            previous
228        }
229        // NoPreviousFail is unreachable here (previous is Some); Completed
230        // falls through to the chained failure.
231        Ok(_) => failure,
232    }
233}
234
235impl tower::Service<Exchange> for DoTryService {
236    type Response = Exchange;
237    type Error = CamelError;
238    type Future = std::pin::Pin<
239        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
240    >;
241
242    fn poll_ready(
243        &mut self,
244        _cx: &mut std::task::Context<'_>,
245    ) -> std::task::Poll<Result<(), Self::Error>> {
246        std::task::Poll::Ready(Ok(()))
247    }
248
249    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
250        // Clear stale CamelExceptionHandled marker from prior handlers in the same route.
251        // Exchange::clear_error() does NOT touch HANDLED, so direct map access is required.
252        exchange.properties.remove(PROPERTY_EXCEPTION_HANDLED);
253
254        let try_steps = self.try_steps.clone();
255        let catch_clauses = self.catch_clauses.clone();
256        let finally_steps = self.finally_steps.clone();
257        let finally_on_when = self.finally_on_when.clone();
258
259        Box::pin(async move {
260            let try_result = run_pipeline(try_steps, exchange).await;
261            match try_result {
262                Ok(ex) => match run_finally(finally_steps, finally_on_when, ex, None).await {
263                    Ok(FinallyOutcome::Completed(ex)) => Ok(ex),
264                    Ok(FinallyOutcome::NoPreviousFail(fin)) => {
265                        tracing::warn!(error = %fin, "doFinally threw");
266                        Err(fin)
267                    }
268                    // Unreachable: the try-Ok flow passes no previous error.
269                    Ok(FinallyOutcome::Restore { previous, .. }) => Err(previous),
270                    // Failed finally `on_when` predicate: the typed error fails
271                    // the scope.
272                    Err(pred_err) => Err(pred_err),
273                },
274                Err(failed) => {
275                    let (failed_ex, original_err) = *failed;
276                    let mut ex = failed_ex;
277                    ex.set_error(original_err.clone());
278
279                    for clause in catch_clauses {
280                        let CatchClause {
281                            matcher,
282                            on_when,
283                            steps,
284                            disposition,
285                        } = clause;
286                        let matched = match matcher.matches(&original_err, &ex).await {
287                            Ok(m) => m,
288                            // Failed catch `when` predicate: replaces the original
289                            // error (preserved as `cause`), fails the scope after
290                            // finally runs.
291                            Err(pred_err) => {
292                                let chained = chain_predicate_error(pred_err, original_err);
293                                return Err(run_finally_for_failure(
294                                    finally_steps,
295                                    finally_on_when,
296                                    ex,
297                                    chained,
298                                )
299                                .await);
300                            }
301                        };
302                        if !matched {
303                            continue;
304                        }
305                        if let Some(ref on_when) = on_when {
306                            match on_when.matches(&ex).await {
307                                Ok(true) => {}
308                                Ok(false) => continue,
309                                // Failed catch `on_when` predicate: same envelope
310                                // as a failed `when` predicate.
311                                Err(pred_err) => {
312                                    let chained = chain_predicate_error(pred_err, original_err);
313                                    return Err(run_finally_for_failure(
314                                        finally_steps,
315                                        finally_on_when,
316                                        ex,
317                                        chained,
318                                    )
319                                    .await);
320                                }
321                            }
322                        }
323
324                        let catch_result = run_pipeline(steps, ex.branch_clone()).await;
325
326                        return match catch_result {
327                            Ok(ok_ex) => {
328                                // Determine previous-error threading based on disposition.
329                                // IMPORTANT: do NOT call handle_error() before run_finally() —
330                                // handle_error() calls clear_error() which removes
331                                // PROPERTY_EXCEPTION_MESSAGE/KIND/CAUGHT, preventing finally
332                                // steps from inspecting the caught exception.
333                                //
334                                // disposition semantics (ADR-0019 strict):
335                                //   Handled    -> catch output is final, no propagation
336                                //   Propagate  -> catch ran for side-effects, original propagates
337                                //   Continued  -> rejected at parse time (defensive: treat as
338                                //                 Propagate + log if we ever reach runtime)
339                                let prev = match disposition {
340                                    ExceptionDisposition::Handled => None,
341                                    ExceptionDisposition::Continued => {
342                                        tracing::warn!(
343                                            "ExceptionDisposition::Continued reached doTry runtime; \
344                                             treating as Propagate. Should have been rejected at parse time."
345                                        );
346                                        Some(original_err.clone())
347                                    }
348                                    // Propagate and any future variant thread the original error.
349                                    _ => Some(original_err.clone()),
350                                };
351                                let mut ex = match run_finally(
352                                    finally_steps.clone(),
353                                    finally_on_when.clone(),
354                                    ok_ex,
355                                    prev,
356                                )
357                                .await
358                                {
359                                    Ok(FinallyOutcome::Completed(ex)) => ex,
360                                    Ok(FinallyOutcome::NoPreviousFail(fin)) => {
361                                        tracing::warn!(error = %fin, "doFinally threw");
362                                        return Err(fin);
363                                    }
364                                    Ok(FinallyOutcome::Restore {
365                                        previous,
366                                        finally_err,
367                                    }) => {
368                                        tracing::warn!(
369                                            finally_error = %finally_err,
370                                            previous_error = %previous,
371                                            "doFinally threw; restoring previous exception (Camel parity)"
372                                        );
373                                        return Err(previous);
374                                    }
375                                    // Failed finally `on_when` predicate: the typed
376                                    // error fails the scope.
377                                    Err(pred_err) => return Err(pred_err),
378                                };
379                                // AFTER finally has run (and had access to exception props),
380                                // apply handle_error() for Handled disposition to clear the
381                                // error state and set CamelExceptionHandled=true marker.
382                                if matches!(disposition, ExceptionDisposition::Handled) {
383                                    ex.handle_error();
384                                }
385                                match disposition {
386                                    ExceptionDisposition::Handled => Ok(ex),
387                                    _ => Err(original_err),
388                                }
389                            }
390                            Err(failed) => {
391                                // Catch threw. Sealed failure envelope (bd rc-zgbqq):
392                                // the catch error stays the main error (returned
393                                // below and threaded into finally); the original is
394                                // surfaced through the unconditional warn record
395                                // and, best-effort, the active span. The event
396                                // lands on the current span when one is entered —
397                                // no new span, no second event.
398                                let (catch_ex, catch_err) = *failed;
399                                tracing::warn!(
400                                    original_error = %original_err,
401                                    catch_error = %catch_err,
402                                    "do_try catch block failed; catch error supersedes original"
403                                );
404                                record_span_error(&catch_err);
405                                // Run finally with previous=catch_err. Per Camel
406                                // parity, if finally itself throws, catch_err is
407                                // restored.
408                                let outcome = run_finally(
409                                    finally_steps.clone(),
410                                    finally_on_when.clone(),
411                                    catch_ex,
412                                    Some(catch_err.clone()),
413                                )
414                                .await;
415                                match outcome {
416                                    Err(pred_err) => return Err(pred_err),
417                                    Ok(FinallyOutcome::Restore {
418                                        previous,
419                                        finally_err,
420                                    }) => {
421                                        tracing::warn!(
422                                            catch_error = %previous,
423                                            finally_error = %finally_err,
424                                            "doFinally threw after failed catch; restoring catch error"
425                                        );
426                                        return Err(previous);
427                                    }
428                                    Ok(_) => {}
429                                }
430                                Err(catch_err)
431                            }
432                        };
433                    }
434
435                    // No catch matched. Run finally with previous=original. Propagate original.
436                    match run_finally(
437                        finally_steps,
438                        finally_on_when,
439                        ex,
440                        Some(original_err.clone()),
441                    )
442                    .await
443                    {
444                        Ok(FinallyOutcome::Restore {
445                            previous,
446                            finally_err,
447                        }) => {
448                            tracing::warn!(
449                                finally_error = %finally_err,
450                                previous_error = %previous,
451                                "doFinally threw; restoring previous exception (Camel parity)"
452                            );
453                            Err(previous)
454                        }
455                        Ok(_) => Err(original_err),
456                        Err(pred_err) => Err(pred_err),
457                    }
458                }
459            }
460        })
461    }
462}
463
464// ── DoTrySegment (ADR-0025 OutcomePipeline) ──────────────────────────────
465
466/// Compilable segment for a `doCatch` clause within a `DoTrySegment`.
467///
468/// `disposition` controls outcome when the catch body completes:
469#[cfg(test)]
470mod tests {
471    use super::*;
472    use crate::test_log_capture::{capture_debugs_with_span_records, captured_field, record_field};
473    use camel_api::{BoxProcessor, BoxProcessorExt};
474    use std::sync::Arc;
475    use std::sync::atomic::{AtomicU32, Ordering};
476
477    fn passthrough() -> BoxProcessor {
478        BoxProcessor::from_fn(move |ex| Box::pin(async move { Ok(ex) }))
479    }
480
481    fn record_call(flag: Arc<AtomicU32>) -> BoxProcessor {
482        BoxProcessor::from_fn(move |ex| {
483            let f = flag.clone();
484            Box::pin(async move {
485                f.fetch_add(1, Ordering::SeqCst);
486                Ok(ex)
487            })
488        })
489    }
490
491    fn always_fail(err: CamelError) -> BoxProcessor {
492        BoxProcessor::from_fn(move |_ex| {
493            let e = err.clone();
494            Box::pin(async move { Err(e) })
495        })
496    }
497
498    #[tokio::test]
499    async fn happy_path_try_succeeds_finally_runs() {
500        let finally_flag = Arc::new(AtomicU32::new(0));
501        let mut svc = DoTryService::new(vec![passthrough()]);
502        svc.finally_steps = vec![record_call(finally_flag.clone())];
503
504        let mut boxed = BoxProcessor::new(svc);
505        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
506        assert!(result.is_ok());
507        assert_eq!(finally_flag.load(Ordering::SeqCst), 1);
508    }
509
510    #[tokio::test]
511    async fn catch_by_variant_handled_returns_ok() {
512        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
513        let mut svc = DoTryService::new(vec![try_step]);
514        svc.catch_clauses.push(CatchClause {
515            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
516            on_when: None,
517            steps: vec![passthrough()],
518            disposition: ExceptionDisposition::Handled,
519        });
520
521        let mut boxed = BoxProcessor::new(svc);
522        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
523        assert!(result.is_ok(), "Handled must return Ok");
524        let ex = result.unwrap();
525        assert_eq!(
526            ex.properties.get(PROPERTY_EXCEPTION_HANDLED),
527            Some(&camel_api::Value::Bool(true)),
528            "CamelExceptionHandled must be set via handle_error()"
529        );
530    }
531
532    #[tokio::test]
533    async fn catch_by_variant_propagate_runs_side_effects_and_rethrows() {
534        let original = CamelError::ProcessorError("boom".into());
535        let try_step = always_fail(original.clone());
536        let side_effect = Arc::new(AtomicU32::new(0));
537        let catch_step = record_call(side_effect.clone());
538        let mut svc = DoTryService::new(vec![try_step]);
539        svc.catch_clauses.push(CatchClause {
540            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
541            on_when: None,
542            steps: vec![catch_step],
543            disposition: ExceptionDisposition::Propagate,
544        });
545
546        let mut boxed = BoxProcessor::new(svc);
547        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
548        assert!(result.is_err(), "Propagate must rethrow original");
549        assert!(matches!(result.unwrap_err(), CamelError::ProcessorError(_)));
550        assert_eq!(
551            side_effect.load(Ordering::SeqCst),
552            1,
553            "catch branch must have run for side-effects"
554        );
555    }
556
557    #[tokio::test]
558    async fn catch_by_predicate_matches_via_exception_kind() {
559        let try_step = always_fail(CamelError::Io("disk full".into()));
560        let predicate = FilterPredicate::new(|ex: &Exchange| {
561            ex.properties
562                .get(camel_api::exchange::PROPERTY_EXCEPTION_KIND)
563                .map(|v| matches!(v, camel_api::Value::String(s) if s == "io"))
564                .unwrap_or(false)
565        });
566        let mut svc = DoTryService::new(vec![try_step]);
567        svc.catch_clauses.push(CatchClause {
568            matcher: CatchMatcher::Predicate(PredicateSource::Sync(predicate)),
569            on_when: None,
570            steps: vec![passthrough()],
571            disposition: ExceptionDisposition::Handled,
572        });
573
574        let mut boxed = BoxProcessor::new(svc);
575        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
576        assert!(
577            result.is_ok(),
578            "Predicate matcher must catch the error and Handled must return Ok"
579        );
580    }
581
582    #[tokio::test]
583    async fn on_when_filters_clause_and_next_evaluated() {
584        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
585        let first_call = Arc::new(AtomicU32::new(0));
586        let second_call = Arc::new(AtomicU32::new(0));
587
588        let mut svc = DoTryService::new(vec![try_step]);
589        svc.catch_clauses.push(CatchClause {
590            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
591            on_when: Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false))),
592            steps: vec![record_call(first_call.clone())],
593            disposition: ExceptionDisposition::Handled,
594        });
595        svc.catch_clauses.push(CatchClause {
596            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
597            on_when: None,
598            steps: vec![record_call(second_call.clone())],
599            disposition: ExceptionDisposition::Handled,
600        });
601
602        let mut boxed = BoxProcessor::new(svc);
603        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
604        assert_eq!(first_call.load(Ordering::SeqCst), 0);
605        assert_eq!(second_call.load(Ordering::SeqCst), 1);
606    }
607
608    #[tokio::test]
609    async fn first_match_wins_subsequent_clauses_not_evaluated() {
610        let try_step = always_fail(CamelError::Io("err".into()));
611        let first_call = Arc::new(AtomicU32::new(0));
612        let second_call = Arc::new(AtomicU32::new(0));
613
614        let mut svc = DoTryService::new(vec![try_step]);
615        svc.catch_clauses.push(CatchClause {
616            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
617            on_when: None,
618            steps: vec![record_call(first_call.clone())],
619            disposition: ExceptionDisposition::Handled,
620        });
621        svc.catch_clauses.push(CatchClause {
622            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
623            on_when: None,
624            steps: vec![record_call(second_call.clone())],
625            disposition: ExceptionDisposition::Handled,
626        });
627
628        let mut boxed = BoxProcessor::new(svc);
629        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
630        assert_eq!(first_call.load(Ordering::SeqCst), 1);
631        assert_eq!(second_call.load(Ordering::SeqCst), 0);
632    }
633
634    #[tokio::test]
635    async fn no_clause_matches_propagates_original() {
636        let try_step = always_fail(CamelError::CircuitOpen("cb".into()));
637        let mut svc = DoTryService::new(vec![try_step]);
638        svc.catch_clauses.push(CatchClause {
639            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
640            on_when: None,
641            steps: vec![passthrough()],
642            disposition: ExceptionDisposition::Handled,
643        });
644
645        let mut boxed = BoxProcessor::new(svc);
646        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
647        assert!(result.is_err());
648        assert!(matches!(result.unwrap_err(), CamelError::CircuitOpen(_)));
649    }
650
651    #[tokio::test]
652    async fn catch_branch_throws_new_error_wins() {
653        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
654        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
655        let mut svc = DoTryService::new(vec![try_step]);
656        svc.catch_clauses.push(CatchClause {
657            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
658            on_when: None,
659            steps: vec![catch_step],
660            disposition: ExceptionDisposition::Handled,
661        });
662
663        let mut boxed = BoxProcessor::new(svc);
664        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
665        assert!(result.is_err());
666        assert!(matches!(result.unwrap_err(), CamelError::Io(_)));
667    }
668
669    #[tokio::test]
670    async fn finally_throws_with_no_previous_error_propagates_finally_error() {
671        let finally_step = always_fail(CamelError::Config("fin".into()));
672        let mut svc = DoTryService::new(vec![passthrough()]);
673        svc.finally_steps = vec![finally_step];
674
675        let mut boxed = BoxProcessor::new(svc);
676        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
677        assert!(result.is_err());
678        assert!(matches!(result.unwrap_err(), CamelError::Config(_)));
679    }
680
681    #[tokio::test]
682    async fn finally_throws_with_previous_error_restores_previous() {
683        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
684        let finally_step = always_fail(CamelError::Config("fin".into()));
685        let mut svc = DoTryService::new(vec![try_step]);
686        svc.finally_steps = vec![finally_step];
687
688        let mut boxed = BoxProcessor::new(svc);
689        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
690        assert!(result.is_err());
691        assert!(
692            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
693            "previous error must be restored when finally throws (Camel parity)"
694        );
695    }
696
697    #[tokio::test]
698    async fn finally_on_when_false_skips_finally() {
699        let finally_call = Arc::new(AtomicU32::new(0));
700        let mut svc = DoTryService::new(vec![passthrough()]);
701        svc.finally_steps = vec![record_call(finally_call.clone())];
702        svc.finally_on_when = Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false)));
703
704        let mut boxed = BoxProcessor::new(svc);
705        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
706        assert_eq!(finally_call.load(Ordering::SeqCst), 0);
707    }
708
709    #[tokio::test]
710    async fn stale_handled_marker_cleared_on_entry() {
711        let mut ex = Exchange::default();
712        ex.set_property(PROPERTY_EXCEPTION_HANDLED, camel_api::Value::Bool(true));
713        let svc = DoTryService::new(vec![passthrough()]);
714        let mut boxed = BoxProcessor::new(svc);
715        let result = boxed.ready().await.unwrap().call(ex).await;
716        let ex = result.unwrap();
717        assert!(
718            !ex.properties.contains_key(PROPERTY_EXCEPTION_HANDLED),
719            "stale CamelExceptionHandled must be cleared on entry"
720        );
721    }
722
723    #[tokio::test]
724    async fn nested_do_try_inner_catch_does_not_leak_to_outer() {
725        let inner = {
726            let try_step = always_fail(CamelError::Io("inner".into()));
727            let mut d = DoTryService::new(vec![try_step]);
728            d.catch_clauses.push(CatchClause {
729                matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
730                on_when: None,
731                steps: vec![passthrough()],
732                disposition: ExceptionDisposition::Handled,
733            });
734            BoxProcessor::new(d)
735        };
736        let mut outer = DoTryService::new(vec![inner]);
737        outer.catch_clauses.push(CatchClause {
738            matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
739            on_when: None,
740            steps: vec![passthrough()],
741            disposition: ExceptionDisposition::Handled,
742        });
743
744        let mut boxed = BoxProcessor::new(outer);
745        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
746        assert!(
747            result.is_ok(),
748            "outer must see Ok because inner handled its own error"
749        );
750    }
751
752    #[tokio::test]
753    async fn catch_all_only_fires_when_no_specific_clause_matches() {
754        let try_step = always_fail(CamelError::Io("err".into()));
755        let processor_call = Arc::new(AtomicU32::new(0));
756        let catch_all_call = Arc::new(AtomicU32::new(0));
757
758        let mut svc = DoTryService::new(vec![try_step]);
759        // First clause (specific) targets ProcessorError — won't match Io error.
760        svc.catch_clauses.push(CatchClause {
761            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
762            on_when: None,
763            steps: vec![record_call(processor_call.clone())],
764            disposition: ExceptionDisposition::Handled,
765        });
766        // Second clause is the catch-all — should fire.
767        svc.catch_clauses.push(CatchClause {
768            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
769            on_when: None,
770            steps: vec![record_call(catch_all_call.clone())],
771            disposition: ExceptionDisposition::Handled,
772        });
773
774        let mut boxed = BoxProcessor::new(svc);
775        let _ = boxed.ready().await.unwrap().call(Exchange::default()).await;
776        assert_eq!(
777            processor_call.load(Ordering::SeqCst),
778            0,
779            "specific ProcessorError clause must not fire on Io error"
780        );
781        assert_eq!(
782            catch_all_call.load(Ordering::SeqCst),
783            1,
784            "catch-all clause must fire when no specific clause matches"
785        );
786    }
787
788    #[tokio::test]
789    async fn catch_throws_with_finally_runs_finally_and_propagates_catch_err() {
790        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
791        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
792        let finally_flag = Arc::new(AtomicU32::new(0));
793        let finally_step = record_call(finally_flag.clone());
794
795        let mut svc = DoTryService::new(vec![try_step]);
796        svc.catch_clauses.push(CatchClause {
797            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
798            on_when: None,
799            steps: vec![catch_step],
800            disposition: ExceptionDisposition::Handled,
801        });
802        svc.finally_steps = vec![finally_step];
803
804        let mut boxed = BoxProcessor::new(svc);
805        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
806
807        assert!(result.is_err());
808        assert!(
809            matches!(result.unwrap_err(), CamelError::Io(_)),
810            "catch_err must propagate (not original ProcessorError)"
811        );
812        assert_eq!(
813            finally_flag.load(Ordering::SeqCst),
814            1,
815            "doFinally must run even when catch throws"
816        );
817    }
818
819    #[tokio::test]
820    async fn catch_throws_and_finally_throws_restores_catch_err() {
821        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
822        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
823        let finally_step = always_fail(CamelError::Config("fin-fail".into()));
824
825        let mut svc = DoTryService::new(vec![try_step]);
826        svc.catch_clauses.push(CatchClause {
827            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
828            on_when: None,
829            steps: vec![catch_step],
830            disposition: ExceptionDisposition::Handled,
831        });
832        svc.finally_steps = vec![finally_step];
833
834        let mut boxed = BoxProcessor::new(svc);
835        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
836
837        assert!(result.is_err());
838        assert!(
839            matches!(result.unwrap_err(), CamelError::Io(_)),
840            "catch_err (Io) must be restored over finally_err (Config) per Camel parity"
841        );
842    }
843
844    #[tokio::test]
845    async fn finally_on_when_false_with_previous_error_still_propagates_original() {
846        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
847        let finally_flag = Arc::new(AtomicU32::new(0));
848        let finally_step = record_call(finally_flag.clone());
849
850        let mut svc = DoTryService::new(vec![try_step]);
851        // No catch clauses → original error stays.
852        svc.finally_steps = vec![finally_step];
853        svc.finally_on_when = Some(PredicateSource::Sync(FilterPredicate::new(|_ex| false)));
854
855        let mut boxed = BoxProcessor::new(svc);
856        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
857
858        assert!(result.is_err());
859        assert!(
860            matches!(result.unwrap_err(), CamelError::ProcessorError(_)),
861            "original error must propagate even when finally_on_when skips finally"
862        );
863        assert_eq!(
864            finally_flag.load(Ordering::SeqCst),
865            0,
866            "doFinally must NOT run when on_when returns false"
867        );
868    }
869
870    fn catch_fails_service() -> BoxProcessor {
871        let try_step = always_fail(CamelError::ProcessorError("orig-lost".into()));
872        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
873        let mut svc = DoTryService::new(vec![try_step]);
874        svc.catch_clauses.push(CatchClause {
875            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
876            on_when: None,
877            steps: vec![catch_step],
878            disposition: ExceptionDisposition::Handled,
879        });
880        BoxProcessor::new(svc)
881    }
882
883    #[tokio::test]
884    async fn catch_throws_under_propagate_disposition_returns_catch_err() {
885        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
886        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
887        let mut svc = DoTryService::new(vec![try_step]);
888        svc.catch_clauses.push(CatchClause {
889            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
890            on_when: None,
891            steps: vec![catch_step],
892            disposition: ExceptionDisposition::Propagate,
893        });
894
895        let mut boxed = BoxProcessor::new(svc);
896        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
897        assert!(
898            matches!(result, Err(CamelError::Io(_))),
899            "Propagate-disposition catch failure must return the catch error, \
900             same envelope as Handled, got: {result:?}"
901        );
902    }
903
904    #[test]
905    fn catch_throws_logs_original_and_catch_error() {
906        let mut boxed = catch_fails_service();
907        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
908            tokio::runtime::Builder::new_current_thread()
909                .enable_all()
910                .build()
911                .expect("current-thread runtime")
912                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
913        });
914
915        assert!(
916            matches!(result, Err(CamelError::Io(_))),
917            "catch failure must return Err(Io), got: {result:?}"
918        );
919        // Structured field lookup on the envelope record, not message
920        // formatting.
921        let original = record_field(&captured, "do_try catch block failed", "original_error")
922            .unwrap_or_else(|| {
923                panic!("envelope record missing original_error field, captured: {captured:?}")
924            });
925        let catch = record_field(&captured, "do_try catch block failed", "catch_error")
926            .unwrap_or_else(|| {
927                panic!("envelope record missing catch_error field, captured: {captured:?}")
928            });
929        assert!(
930            original.contains("orig-lost"),
931            "original_error must carry the original error, got: {original}"
932        );
933        assert!(
934            catch.contains("catch-fail"),
935            "catch_error must carry the catch error, got: {catch}"
936        );
937    }
938
939    #[test]
940    fn catch_throws_marks_span_error_and_event() {
941        let mut boxed = catch_fails_service();
942        let (result, captured, span_records) = capture_debugs_with_span_records(|| {
943            // Declared `error` field so record_span_error's record lands.
944            let span = tracing::info_span!("dotry_test", error = tracing::field::Empty);
945            let _guard = span.enter();
946            tokio::runtime::Builder::new_current_thread()
947                .enable_all()
948                .build()
949                .expect("current-thread runtime")
950                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
951        });
952
953        assert!(
954            matches!(result, Err(CamelError::Io(_))),
955            "catch failure must return Err(Io), got: {result:?}"
956        );
957        // The span recorded the `error` field with the CATCH error.
958        // Structured field lookup: substring `error=` would also match
959        // `original_error=` suffixes.
960        assert!(
961            span_records.iter().any(|line| {
962                captured_field(line, "error").is_some_and(|v| v.contains("catch-fail"))
963            }),
964            "expected span error record carrying the catch error, span records: {span_records:?}"
965        );
966        // The WARN event still carries original_error while a span is active.
967        let original = record_field(
968            &captured,
969            "do_try catch block failed",
970            "original_error",
971        )
972        .unwrap_or_else(|| {
973            panic!(
974                "envelope record missing original_error field under active span, captured: {captured:?}"
975            )
976        });
977        assert!(
978            original.contains("orig-lost"),
979            "original_error must carry the original error under an active span, got: {original}"
980        );
981    }
982
983    #[test]
984    fn catch_and_finally_throw_logs_finally_error() {
985        let try_step = always_fail(CamelError::ProcessorError("orig".into()));
986        let catch_step = always_fail(CamelError::Io("catch-fail".into()));
987        let finally_step = always_fail(CamelError::Config("fin-fail".into()));
988        let mut svc = DoTryService::new(vec![try_step]);
989        svc.catch_clauses.push(CatchClause {
990            matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
991            on_when: None,
992            steps: vec![catch_step],
993            disposition: ExceptionDisposition::Handled,
994        });
995        svc.finally_steps = vec![finally_step];
996
997        let mut boxed = BoxProcessor::new(svc);
998        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
999            tokio::runtime::Builder::new_current_thread()
1000                .enable_all()
1001                .build()
1002                .expect("current-thread runtime")
1003                .block_on(async { boxed.ready().await.unwrap().call(Exchange::default()).await })
1004        });
1005
1006        assert!(
1007            matches!(result, Err(CamelError::Io(_))),
1008            "catch error must be restored over finally error, got: {result:?}"
1009        );
1010        let catch = record_field(
1011            &captured,
1012            "doFinally threw after failed catch; restoring catch error",
1013            "catch_error",
1014        )
1015        .unwrap_or_else(|| {
1016            panic!("catch-failed restore record missing catch_error field, captured: {captured:?}")
1017        });
1018        let finally = record_field(
1019            &captured,
1020            "doFinally threw after failed catch; restoring catch error",
1021            "finally_error",
1022        )
1023        .unwrap_or_else(|| {
1024            panic!(
1025                "catch-failed restore record missing finally_error field, captured: {captured:?}"
1026            )
1027        });
1028        assert!(
1029            catch.contains("catch-fail"),
1030            "catch_error must carry the catch error, got: {catch}"
1031        );
1032        assert!(
1033            finally.contains("fin-fail"),
1034            "finally_error must carry the finally error, got: {finally}"
1035        );
1036    }
1037
1038    // ── Fallible predicate path (language-value-boundary task 1.4) ──
1039
1040    use camel_api::{ExpressionErrorClass, FilterPredicate, PredicateSource};
1041
1042    fn expression_failed() -> CamelError {
1043        CamelError::ExpressionFailed {
1044            language: "rhai".to_string(),
1045            route_id: "r1".to_string(),
1046            step_id: "step#0".to_string(),
1047            verb: "when".to_string(),
1048            class: ExpressionErrorClass::Runtime,
1049            position: None,
1050            conversion: None,
1051            cause: None,
1052        }
1053    }
1054
1055    fn async_err_predicate(err: CamelError) -> PredicateSource {
1056        PredicateSource::Async(Arc::new(move |_: &Exchange| {
1057            let err = err.clone();
1058            Box::pin(async move { Err(err) }) as camel_api::BoxBoolFuture
1059        }))
1060    }
1061
1062    #[tokio::test]
1063    async fn catch_when_predicate_error_chains_original() {
1064        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
1065        let mut svc = DoTryService::new(vec![try_step]);
1066        svc.catch_clauses.push(CatchClause {
1067            matcher: CatchMatcher::Predicate(async_err_predicate(expression_failed())),
1068            on_when: None,
1069            steps: vec![passthrough()],
1070            disposition: ExceptionDisposition::Handled,
1071        });
1072
1073        let mut boxed = BoxProcessor::new(svc);
1074        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
1075        match result {
1076            Err(CamelError::ExpressionFailed {
1077                cause: Some(cause), ..
1078            }) => {
1079                assert!(
1080                    cause.to_string().contains("boom"),
1081                    "cause must carry the original error, got: {cause}"
1082                );
1083            }
1084            other => panic!("expected Err(ExpressionFailed) with chained cause, got {other:?}"),
1085        }
1086    }
1087
1088    #[tokio::test]
1089    async fn catch_on_when_predicate_error_chains_original() {
1090        let try_step = always_fail(CamelError::ProcessorError("boom".into()));
1091        let mut svc = DoTryService::new(vec![try_step]);
1092        svc.catch_clauses.push(CatchClause {
1093            matcher: CatchMatcher::ByVariant(vec!["*".into()]),
1094            on_when: Some(async_err_predicate(expression_failed())),
1095            steps: vec![passthrough()],
1096            disposition: ExceptionDisposition::Handled,
1097        });
1098
1099        let mut boxed = BoxProcessor::new(svc);
1100        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
1101        match result {
1102            Err(CamelError::ExpressionFailed {
1103                cause: Some(cause), ..
1104            }) => {
1105                assert!(
1106                    cause.to_string().contains("boom"),
1107                    "cause must carry the original error, got: {cause}"
1108                );
1109            }
1110            other => panic!("expected Err(ExpressionFailed) with chained cause, got {other:?}"),
1111        }
1112    }
1113
1114    #[tokio::test]
1115    async fn finally_on_when_predicate_error_fails() {
1116        let mut svc = DoTryService::new(vec![passthrough()]);
1117        svc.finally_steps = vec![passthrough()];
1118        svc.finally_on_when = Some(async_err_predicate(expression_failed()));
1119
1120        let mut boxed = BoxProcessor::new(svc);
1121        let result = boxed.ready().await.unwrap().call(Exchange::default()).await;
1122        match result {
1123            Err(CamelError::ExpressionFailed { .. }) => {}
1124            other => panic!("expected Err(ExpressionFailed), got {other:?}"),
1125        }
1126    }
1127
1128    // ── Mapper-generated cause replacement (task 1.3) ──
1129
1130    use camel_language_api::{EvalMeta, LanguageError, to_expression_failed};
1131
1132    fn caller_diagnostic_failed(cause: CamelError) -> CamelError {
1133        CamelError::ExpressionFailed {
1134            language: "rhai".to_string(),
1135            route_id: "r1".to_string(),
1136            step_id: "s1".to_string(),
1137            verb: "when".to_string(),
1138            class: ExpressionErrorClass::Runtime,
1139            position: None,
1140            conversion: None,
1141            cause: Some(Box::new(cause)),
1142        }
1143    }
1144
1145    #[test]
1146    fn chain_replaces_mapper_generated_cause() {
1147        let meta = EvalMeta {
1148            language: "xpath".into(),
1149            route_id: "r1".into(),
1150            step_id: "s1".into(),
1151            verb: "when".into(),
1152            target: None,
1153        };
1154        let mapper_err = to_expression_failed(
1155            LanguageError::EvalFailure {
1156                class: ExpressionErrorClass::Parse,
1157                position: None,
1158                detail: Some("body is not valid XML".into()),
1159            },
1160            &meta,
1161        );
1162
1163        let result = chain_predicate_error(mapper_err, CamelError::ProcessorError("boom".into()));
1164
1165        match result {
1166            CamelError::ExpressionFailed {
1167                cause: Some(cause), ..
1168            } => {
1169                assert!(
1170                    cause.to_string().contains("boom"),
1171                    "mapper diagnostic cause must be replaced by the original, got: {cause}"
1172                );
1173            }
1174            other => panic!("expected ExpressionFailed with chained cause, got {other:?}"),
1175        }
1176    }
1177
1178    #[test]
1179    fn chain_preserves_caller_cause() {
1180        let predicate_err =
1181            caller_diagnostic_failed(CamelError::ProcessorError("caller diagnostic".into()));
1182
1183        let result =
1184            chain_predicate_error(predicate_err, CamelError::ProcessorError("boom".into()));
1185
1186        match result {
1187            CamelError::ExpressionFailed {
1188                cause: Some(cause), ..
1189            } => {
1190                let rendered = cause.to_string();
1191                assert!(
1192                    rendered.contains("caller diagnostic"),
1193                    "caller cause must pass through untouched, got: {rendered}"
1194                );
1195                assert!(
1196                    !rendered.contains("boom"),
1197                    "caller cause must not be replaced, got: {rendered}"
1198                );
1199            }
1200            other => panic!("expected ExpressionFailed with caller cause, got {other:?}"),
1201        }
1202    }
1203
1204    #[test]
1205    fn chain_preserves_caller_source_bearing_cause() {
1206        let predicate_err = caller_diagnostic_failed(CamelError::ProcessorErrorWithSource(
1207            "caller diagnostic".into(),
1208            Arc::new(std::io::Error::other("caller source")),
1209        ));
1210
1211        let result =
1212            chain_predicate_error(predicate_err, CamelError::ProcessorError("boom".into()));
1213
1214        match result {
1215            CamelError::ExpressionFailed {
1216                cause: Some(cause), ..
1217            } => {
1218                let rendered = cause.to_string();
1219                assert!(
1220                    rendered.contains("caller diagnostic"),
1221                    "caller cause must pass through untouched, got: {rendered}"
1222                );
1223                let source = std::error::Error::source(cause.as_ref())
1224                    .unwrap_or_else(|| panic!("caller source must survive, cause: {cause:?}"));
1225                assert!(
1226                    source.to_string().contains("caller source"),
1227                    "caller source must survive untouched, got: {source}"
1228                );
1229            }
1230            other => panic!("expected ExpressionFailed with caller cause, got {other:?}"),
1231        }
1232    }
1233}
camel_processor/do_try.rs

camel_processor/
do_try.rs
