camel_processor/
loop_eip.rs

1use std::future::Future;
2use std::pin::Pin;
3
4pub const CAMEL_LOOP_INDEX: &str = "CamelLoopIndex";
5pub const CAMEL_LOOP_SIZE: &str = "CamelLoopSize";
6
7// ── LoopSegment (ADR-0025 OutcomePipeline) ─────────────────────────────
8
9/// Outcome-aware structural EIP segment for the Loop pattern.
10///
11/// Operates at the `PipelineOutcome` layer so that `Stopped(ex)` from a
12/// sub-step (e.g. Stop EIP) is preserved with the exchange including all
13/// mutations. Supports both Count and While modes.
14///
15/// The segment short-circuits on `PipelineOutcome::Stopped` or `Failed`.
16pub struct LoopSegment {
17    pub config: camel_api::loop_eip::LoopConfig,
18    pub body: camel_api::OutcomeSegment,
19}
20
21impl Clone for LoopSegment {
22    fn clone(&self) -> Self {
23        Self {
24            config: self.config.clone(),
25            body: self.body.clone(),
26        }
27    }
28}
29
30impl camel_api::OutcomePipeline for LoopSegment {
31    fn clone_box(&self) -> Box<dyn camel_api::OutcomePipeline> {
32        Box::new(self.clone())
33    }
34
35    fn run<'a>(
36        &'a mut self,
37        exchange: camel_api::Exchange,
38    ) -> Pin<Box<dyn Future<Output = camel_api::PipelineOutcome> + Send + 'a>> {
39        use camel_api::{PipelineOutcome, Value};
40
41        let config = self.config.clone();
42        let body = &mut self.body;
43
44        Box::pin(async move {
45            match config.mode {
46                camel_api::loop_eip::LoopMode::Count(n) => {
47                    let n_clamped = n.min(config.max_iterations);
48                    if n > config.max_iterations {
49                        tracing::warn!(
50                            requested = n,
51                            clamped_to = config.max_iterations,
52                            "LoopMode::Count exceeded max_iterations; clamping"
53                        );
54                    }
55                    let mut ex = exchange;
56                    ex.set_property(CAMEL_LOOP_SIZE, Value::from(n_clamped as u64));
57                    for i in 0..n_clamped {
58                        ex.set_property(CAMEL_LOOP_INDEX, Value::from(i as u64));
59                        match body.run(ex).await {
60                            PipelineOutcome::Completed(next) => {
61                                ex = next;
62                            }
63                            other => return other,
64                        }
65                    }
66                    PipelineOutcome::Completed(ex)
67                }
68                camel_api::loop_eip::LoopMode::While(ref predicate) => {
69                    let mut ex = exchange;
70                    ex.set_property(CAMEL_LOOP_SIZE, Value::from(0u64));
71                    // Every predicate evaluation error propagates as
72                    // `Failed(err)` (never as loop-end Completed), including
73                    // the post-exhaustion safety-guard probe.
74                    let mut i = 0u64;
75                    let mut exhausted = true;
76                    while i < config.max_iterations as u64 {
77                        match predicate.matches(&ex).await {
78                            Ok(true) => {}
79                            Ok(false) => {
80                                exhausted = false;
81                                break;
82                            }
83                            Err(err) => return PipelineOutcome::Failed(err),
84                        }
85                        ex.set_property(CAMEL_LOOP_INDEX, Value::from(i));
86                        match body.run(ex).await {
87                            PipelineOutcome::Completed(next) => {
88                                ex = next;
89                            }
90                            other => return other,
91                        }
92                        i += 1;
93                    }
94                    if exhausted {
95                        match predicate.matches(&ex).await {
96                            Ok(true) => {
97                                tracing::warn!(
98                                    "Loop while-mode hit max_iterations ({}) safety guard. Predicate still true.",
99                                    config.max_iterations
100                                );
101                            }
102                            Ok(false) => {}
103                            Err(err) => return PipelineOutcome::Failed(err),
104                        }
105                    }
106                    PipelineOutcome::Completed(ex)
107                }
108                // Future loop modes: no iteration, complete with the exchange unchanged.
109                _ => PipelineOutcome::Completed(exchange),
110            }
111        })
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use std::sync::atomic::{AtomicUsize, Ordering};
118    use std::sync::{Arc, Mutex};
119
120    use camel_api::loop_eip::{LoopConfig, LoopMode, MAX_LOOP_ITERATIONS};
121    use camel_api::{
122        Body, BoxProcessor, BoxProcessorExt, CamelError, Exchange, ExpressionErrorClass,
123        FilterPredicate, IdentityProcessor, Message, OutcomePipeline, PipelineOutcome,
124        PredicateSource,
125    };
126
127    use super::{CAMEL_LOOP_INDEX, CAMEL_LOOP_SIZE, LoopSegment};
128    use crate::test_processor_segment::processor_segment;
129
130    fn identity_pipeline() -> BoxProcessor {
131        BoxProcessor::new(IdentityProcessor)
132    }
133
134    fn counter_pipeline(counter: Arc<AtomicUsize>) -> BoxProcessor {
135        BoxProcessor::from_fn(move |exchange: Exchange| {
136            let counter = Arc::clone(&counter);
137            Box::pin(async move {
138                counter.fetch_add(1, Ordering::SeqCst);
139                Ok(exchange)
140            })
141        })
142    }
143
144    fn segment(config: LoopConfig, body: BoxProcessor) -> LoopSegment {
145        LoopSegment {
146            config,
147            body: processor_segment(body),
148        }
149    }
150
151    fn completed(outcome: PipelineOutcome) -> Exchange {
152        match outcome {
153            PipelineOutcome::Completed(ex) => ex,
154            other => panic!("expected PipelineOutcome::Completed, got {other:?}"),
155        }
156    }
157
158    #[tokio::test]
159    async fn loop_segment_count_iterates_n_times() {
160        let counter = Arc::new(AtomicUsize::new(0));
161        let config = LoopConfig::new(LoopMode::Count(3));
162        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
163
164        let exchange = Exchange::new(Message::new("test"));
165        let outcome = seg.run(exchange).await;
166
167        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
168        assert_eq!(counter.load(Ordering::SeqCst), 3);
169    }
170
171    #[tokio::test]
172    async fn loop_segment_count_sets_properties() {
173        let seen_indices = Arc::new(Mutex::new(Vec::<u64>::new()));
174        let seen_indices_for_pipeline = Arc::clone(&seen_indices);
175
176        let pipeline = BoxProcessor::from_fn(move |exchange: Exchange| {
177            let seen_indices = Arc::clone(&seen_indices_for_pipeline);
178            Box::pin(async move {
179                if let Some(index) = exchange.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()) {
180                    seen_indices.lock().unwrap().push(index);
181                }
182                Ok(exchange)
183            })
184        });
185
186        let config = LoopConfig::new(LoopMode::Count(3));
187        let mut seg = segment(config, pipeline);
188
189        let exchange = Exchange::new(Message::new("test"));
190        let result = completed(seg.run(exchange).await);
191
192        assert_eq!(*seen_indices.lock().unwrap(), vec![0, 1, 2]);
193        assert_eq!(
194            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
195            Some(3)
196        );
197    }
198
199    #[tokio::test]
200    async fn loop_segment_count_zero_is_noop() {
201        let config = LoopConfig::new(LoopMode::Count(0));
202        let mut seg = segment(config, identity_pipeline());
203
204        let exchange = Exchange::new(Message::new("test"));
205        let result = completed(seg.run(exchange).await);
206
207        assert_eq!(result.input.body.as_text(), Some("test"));
208        assert_eq!(
209            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
210            Some(0)
211        );
212        assert!(result.property(CAMEL_LOOP_INDEX).is_none());
213    }
214
215    #[tokio::test]
216    async fn loop_segment_while_stops_when_predicate_false() {
217        let counter = Arc::new(AtomicUsize::new(0));
218
219        let predicate = FilterPredicate::new(|exchange: &Exchange| {
220            exchange
221                .property("iterations")
222                .and_then(|v| v.as_u64())
223                .unwrap_or(0)
224                < 2
225        });
226
227        let counter_for_pipeline = Arc::clone(&counter);
228        let pipeline = BoxProcessor::from_fn(move |mut exchange: Exchange| {
229            let counter = Arc::clone(&counter_for_pipeline);
230            Box::pin(async move {
231                let current = exchange
232                    .property("iterations")
233                    .and_then(|v| v.as_u64())
234                    .unwrap_or(0);
235                exchange.set_property("iterations", current + 1);
236                counter.fetch_add(1, Ordering::SeqCst);
237                Ok(exchange)
238            })
239        });
240
241        let config = LoopConfig::new(LoopMode::While(PredicateSource::Sync(predicate)));
242        let mut seg = segment(config, pipeline);
243
244        let exchange = Exchange::new(Message::new("test"));
245        let result = completed(seg.run(exchange).await);
246
247        assert_eq!(counter.load(Ordering::SeqCst), 2);
248        assert_eq!(
249            result.property("iterations").and_then(|v| v.as_u64()),
250            Some(2)
251        );
252        assert_eq!(
253            result.property(CAMEL_LOOP_INDEX).and_then(|v| v.as_u64()),
254            Some(1)
255        );
256        assert_eq!(
257            result.property(CAMEL_LOOP_SIZE).and_then(|v| v.as_u64()),
258            Some(0)
259        );
260    }
261
262    #[tokio::test]
263    async fn loop_segment_while_respects_max_iterations() {
264        let counter = Arc::new(AtomicUsize::new(0));
265        let predicate = FilterPredicate::new(|_exchange: &Exchange| true);
266        let config = LoopConfig::new(LoopMode::While(PredicateSource::Sync(predicate)));
267        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
268
269        let exchange = Exchange::new(Message::new("test"));
270        let outcome = seg.run(exchange).await;
271
272        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
273        assert_eq!(counter.load(Ordering::SeqCst), MAX_LOOP_ITERATIONS);
274    }
275
276    #[tokio::test]
277    async fn loop_segment_error_propagation() {
278        let pipeline = BoxProcessor::from_fn(|_exchange: Exchange| {
279            Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
280        });
281
282        let config = LoopConfig::new(LoopMode::Count(3));
283        let mut seg = segment(config, pipeline);
284
285        let exchange = Exchange::new(Message::new("test"));
286        let outcome = seg.run(exchange).await;
287
288        assert!(matches!(
289            outcome,
290            PipelineOutcome::Failed(CamelError::ProcessorError(msg)) if msg == "boom"
291        ));
292    }
293
294    // ── D-M9 Batch 1: LoopMode::Count(n) clamped to MAX_LOOP_ITERATIONS ──
295
296    /// D-M9: a `Count(u32::MAX as usize)` (or any value above
297    /// `MAX_LOOP_ITERATIONS`) is clamped to `MAX_LOOP_ITERATIONS` and
298    /// runs at most that many iterations. Without the clamp, a malicious
299    /// or typo'd `count: 4294967295` would exhaust CPU. The pre-existing
300    /// `Count(3)` test continues to pass — small values are unaffected.
301    #[tokio::test]
302    async fn loop_segment_count_clamped_to_max_iterations() {
303        let counter = Arc::new(AtomicUsize::new(0));
304        let config = LoopConfig::new(LoopMode::Count(usize::MAX));
305        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
306
307        let exchange = Exchange::new(Message::new("test"));
308        let outcome = seg.run(exchange).await;
309
310        assert!(matches!(&outcome, PipelineOutcome::Completed(_)));
311        // Must run exactly MAX_LOOP_ITERATIONS, not u32::MAX iterations.
312        assert_eq!(counter.load(Ordering::SeqCst), MAX_LOOP_ITERATIONS);
313        // The CAMEL_LOOP_SIZE property is set to the *clamped* count.
314        assert_eq!(
315            completed(outcome)
316                .property(CAMEL_LOOP_SIZE)
317                .and_then(|v| v.as_u64()),
318            Some(MAX_LOOP_ITERATIONS as u64)
319        );
320    }
321
322    #[tokio::test]
323    async fn loop_segment_count_with_custom_max_iterations() {
324        let counter = Arc::new(AtomicUsize::new(0));
325        let config = LoopConfig::new(LoopMode::Count(15_000)).with_max_iterations(15_000);
326        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
327        let exchange = Exchange::new(Message::new("test"));
328        let _ = completed(seg.run(exchange).await);
329        assert_eq!(counter.load(Ordering::SeqCst), 15_000);
330    }
331
332    #[tokio::test]
333    async fn loop_segment_while_with_custom_max_iterations() {
334        let counter = Arc::new(AtomicUsize::new(0));
335        let predicate = FilterPredicate::new(|_| true);
336        let config = LoopConfig::new(LoopMode::While(PredicateSource::Sync(predicate)))
337            .with_max_iterations(50);
338        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
339        let exchange = Exchange::new(Message::new("test"));
340        let _ = completed(seg.run(exchange).await);
341        assert_eq!(counter.load(Ordering::SeqCst), 50);
342    }
343
344    #[tokio::test]
345    async fn loop_segment_pipeline_chaining() {
346        let pipeline = BoxProcessor::from_fn(|mut exchange: Exchange| {
347            Box::pin(async move {
348                if let Body::Text(s) = &exchange.input.body {
349                    exchange.input.body = Body::Text(format!("{s}x"));
350                }
351                Ok(exchange)
352            })
353        });
354
355        let config = LoopConfig::new(LoopMode::Count(3));
356        let mut seg = segment(config, pipeline);
357
358        let exchange = Exchange::new(Message::new("start"));
359        let result = completed(seg.run(exchange).await);
360
361        assert_eq!(result.input.body.as_text(), Some("startxxx"));
362    }
363
364    // ── Fallible predicate path (language-value-boundary task 1.4) ──
365
366    fn expression_failed() -> CamelError {
367        CamelError::ExpressionFailed {
368            language: "rhai".to_string(),
369            route_id: "r1".to_string(),
370            step_id: "step#0".to_string(),
371            verb: "loop".to_string(),
372            class: ExpressionErrorClass::Runtime,
373            position: None,
374            conversion: None,
375            cause: None,
376        }
377    }
378
379    fn async_err_predicate(err: CamelError) -> PredicateSource {
380        PredicateSource::Async(Arc::new(move |_: &Exchange| {
381            let err = err.clone();
382            Box::pin(async move { Err(err) }) as camel_api::BoxBoolFuture
383        }))
384    }
385
386    /// Async predicate that returns `Ok(true)` on its first call and
387    /// `Err(expression_failed())` afterwards. Paired with
388    /// `max_iterations = 1` it drives the loop into the exhaustion
389    /// safety-guard probe on the second evaluation.
390    fn exhausted_then_err_predicate(calls: Arc<AtomicUsize>) -> PredicateSource {
391        PredicateSource::Async(Arc::new(move |_: &Exchange| {
392            let call = calls.fetch_add(1, Ordering::SeqCst);
393            let err = expression_failed();
394            Box::pin(async move { if call == 0 { Ok(true) } else { Err(err) } })
395                as camel_api::BoxBoolFuture
396        }))
397    }
398
399    #[tokio::test]
400    async fn loop_segment_while_predicate_error_fails_loop() {
401        let counter = Arc::new(AtomicUsize::new(0));
402        let config = LoopConfig::new(LoopMode::While(async_err_predicate(expression_failed())));
403        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
404
405        let exchange = Exchange::new(Message::new("test"));
406        let outcome = seg.run(exchange).await;
407
408        match outcome {
409            PipelineOutcome::Failed(CamelError::ExpressionFailed { .. }) => {}
410            other => panic!("expected Failed(ExpressionFailed), got {other:?}"),
411        }
412        assert_eq!(
413            counter.load(Ordering::SeqCst),
414            0,
415            "loop body must not run when the while predicate errors"
416        );
417    }
418
419    #[tokio::test]
420    async fn loop_segment_guard_predicate_error_propagates() {
421        let calls = Arc::new(AtomicUsize::new(0));
422        let counter = Arc::new(AtomicUsize::new(0));
423        let config = LoopConfig::new(LoopMode::While(exhausted_then_err_predicate(Arc::clone(
424            &calls,
425        ))))
426        .with_max_iterations(1);
427        let mut seg = segment(config, counter_pipeline(Arc::clone(&counter)));
428
429        let exchange = Exchange::new(Message::new("test"));
430        let outcome = seg.run(exchange).await;
431
432        match outcome {
433            PipelineOutcome::Failed(CamelError::ExpressionFailed { .. }) => {}
434            other => panic!("expected Failed(ExpressionFailed), got {other:?}"),
435        }
436        assert_eq!(
437            counter.load(Ordering::SeqCst),
438            1,
439            "body runs once before the exhaustion guard probe errors"
440        );
441        assert_eq!(
442            calls.load(Ordering::SeqCst),
443            2,
444            "predicate evaluated once in-loop and once by the safety guard"
445        );
446    }
447}
camel_processor/loop_eip.rs

camel_processor/
loop_eip.rs
