camel_api/
aggregator.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::error::{CamelError, ConfigValidationError};
5use crate::exchange::Exchange;
6
7/// Aggregation function — left-fold binary: (accumulated, next) -> merged.
8pub type AggregationFn = Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync>;
9
10/// Strategy for correlating exchanges into aggregation buckets.
11pub enum CorrelationStrategy {
12    /// Correlate by the value of a named header.
13    HeaderName(String),
14    /// Correlate by evaluating an expression using a language registry.
15    Expression { expr: String, language: String },
16    /// Correlate using a custom function.
17    #[allow(clippy::type_complexity)]
18    Fn(Arc<dyn Fn(&Exchange) -> Option<String> + Send + Sync>),
19}
20
21impl Clone for CorrelationStrategy {
22    fn clone(&self) -> Self {
23        match self {
24            CorrelationStrategy::HeaderName(h) => CorrelationStrategy::HeaderName(h.clone()),
25            CorrelationStrategy::Expression { expr, language } => CorrelationStrategy::Expression {
26                expr: expr.clone(),
27                language: language.clone(),
28            },
29            CorrelationStrategy::Fn(f) => CorrelationStrategy::Fn(Arc::clone(f)),
30        }
31    }
32}
33
34impl std::fmt::Debug for CorrelationStrategy {
35    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        match self {
37            CorrelationStrategy::HeaderName(h) => f.debug_tuple("HeaderName").field(h).finish(),
38            CorrelationStrategy::Expression { expr, language } => f
39                .debug_struct("Expression")
40                .field("expr", expr)
41                .field("language", language)
42                .finish(),
43            CorrelationStrategy::Fn(_) => f.write_str("Fn(..)"),
44        }
45    }
46}
47
48/// How to combine collected exchanges into one.
49#[derive(Clone)]
50pub enum AggregationStrategy {
51    /// Collects all bodies into Body::Json([body1, body2, ...]).
52    CollectAll,
53    /// Left-fold: f(f(ex1, ex2), ex3), ...
54    Custom(AggregationFn),
55}
56
57/// When the bucket is considered complete and should be emitted.
58#[derive(Clone)]
59pub enum CompletionCondition {
60    /// Emit when bucket reaches exactly N exchanges.
61    Size(usize),
62    /// Emit when predicate returns true for current bucket.
63    #[allow(clippy::type_complexity)]
64    Predicate(Arc<dyn Fn(&[Exchange]) -> bool + Send + Sync>),
65    /// Emit when the bucket has been inactive for the given duration.
66    Timeout(Duration),
67}
68
69/// Determines how a bucket's completion is evaluated.
70/// `Single` wraps one condition; `Any` completes when the first condition triggers.
71#[derive(Clone)]
72pub enum CompletionMode {
73    Single(CompletionCondition),
74    Any(Vec<CompletionCondition>),
75}
76
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub enum CompletionReason {
79    Size,
80    Predicate,
81    Timeout,
82    Stop,
83}
84
85impl CompletionReason {
86    pub fn as_str(&self) -> &'static str {
87        match self {
88            CompletionReason::Size => "size",
89            CompletionReason::Predicate => "predicate",
90            CompletionReason::Timeout => "timeout",
91            CompletionReason::Stop => "stop",
92        }
93    }
94}
95
96/// Configuration for the Aggregator EIP.
97#[derive(Clone)]
98pub struct AggregatorConfig {
99    /// Name of the header used as correlation key.
100    pub header_name: String,
101    /// When to emit the aggregated exchange.
102    pub completion: CompletionMode,
103    /// Strategy for determining correlation keys.
104    pub correlation: CorrelationStrategy,
105    /// How to combine the bucket into one exchange.
106    pub strategy: AggregationStrategy,
107    /// Maximum number of correlation key buckets (memory protection).
108    /// When limit is reached, new correlation keys are rejected.
109    pub max_buckets: Option<usize>,
110    /// Time-to-live for inactive buckets (memory protection).
111    /// Buckets not updated for this duration are evicted.
112    pub bucket_ttl: Option<Duration>,
113    /// Force-complete all pending buckets when the route is stopped.
114    pub force_completion_on_stop: bool,
115    /// Discard bucket contents on timeout instead of emitting.
116    pub discard_on_timeout: bool,
117    /// Maximum number of concurrently-live per-bucket timeout tasks (DoS cap, R3-M3).
118    /// When the cap is reached, new buckets skip the dedicated timeout spawn and
119    /// rely on `bucket_ttl` eviction (graceful degradation under a key flood).
120    /// Default 1024.
121    pub max_timeout_tasks: usize,
122}
123
124impl AggregatorConfig {
125    /// Start building config with correlation key extracted from the named header.
126    pub fn correlate_by(header: impl Into<String>) -> AggregatorConfigBuilder {
127        let header_name = header.into();
128        AggregatorConfigBuilder {
129            header_name: header_name.clone(),
130            completion: None,
131            correlation: CorrelationStrategy::HeaderName(header_name),
132            strategy: AggregationStrategy::CollectAll,
133            // R3-C1 Batch 1: bounded defaults — the correlation map MUST NOT
134            // grow without a cap. A flood of unique correlation keys is the
135            // remote-OOM vector; the default cap (10_000) is a sane production
136            // ceiling the operator may still override. The TTL (5 minutes) is
137            // the inline-retain + background-sweep eviction window.
138            max_buckets: Some(10_000),
139            bucket_ttl: Some(Duration::from_secs(300)),
140            force_completion_on_stop: false,
141            discard_on_timeout: false,
142            // R3-M3: cap concurrently-live per-bucket timeout tasks.
143            max_timeout_tasks: 1024,
144        }
145    }
146
147    /// Validate that at least one memory-release bound is configured (R3-M2).
148    ///
149    /// At least one of `max_buckets`, a `Timeout` completion condition, or
150    /// `bucket_ttl` MUST be set, otherwise a flood of unique correlation keys
151    /// grows the bucket map without limit (remote-OOM vector).
152    ///
153    /// Additionally, when a `Timeout` completion condition is present,
154    /// `bucket_ttl` MUST also be set. The R3-M3 timeout-task cap may skip
155    /// spawning a dedicated timeout task under flood; without `bucket_ttl`
156    /// there is no fallback eviction path and the bucket leaks until shutdown.
157    /// Requiring `bucket_ttl` whenever Timeout is present makes the cap-skip
158    /// degradation safe by construction.
159    pub fn validate(&self) -> Result<(), CamelError> {
160        let has_timeout = match &self.completion {
161            CompletionMode::Single(CompletionCondition::Timeout(_)) => true,
162            CompletionMode::Any(conds) => conds
163                .iter()
164                .any(|c| matches!(c, CompletionCondition::Timeout(_))),
165            _ => false,
166        };
167        let has_bound = self.max_buckets.is_some() || has_timeout || self.bucket_ttl.is_some();
168        if !has_bound {
169            return Err(CamelError::from(
170                ConfigValidationError::AggregatorMissingMemoryBound,
171            ));
172        }
173        // R3-M3: Timeout completion requires bucket_ttl so the cap-skip
174        // degradation always has an eviction path.
175        if has_timeout && self.bucket_ttl.is_none() {
176            return Err(CamelError::from(
177                ConfigValidationError::AggregatorTimeoutRequiresTtl,
178            ));
179        }
180        Ok(())
181    }
182}
183
184/// Builder for `AggregatorConfig`.
185pub struct AggregatorConfigBuilder {
186    header_name: String,
187    completion: Option<CompletionMode>,
188    correlation: CorrelationStrategy,
189    strategy: AggregationStrategy,
190    max_buckets: Option<usize>,
191    bucket_ttl: Option<Duration>,
192    force_completion_on_stop: bool,
193    discard_on_timeout: bool,
194    max_timeout_tasks: usize,
195}
196
197impl AggregatorConfigBuilder {
198    /// Emit when bucket has N exchanges.
199    pub fn complete_when_size(mut self, n: usize) -> Self {
200        self.completion = Some(CompletionMode::Single(CompletionCondition::Size(n)));
201        self
202    }
203
204    /// Emit when predicate returns true for the current bucket.
205    pub fn complete_when<F>(mut self, predicate: F) -> Self
206    where
207        F: Fn(&[Exchange]) -> bool + Send + Sync + 'static,
208    {
209        self.completion = Some(CompletionMode::Single(CompletionCondition::Predicate(
210            Arc::new(predicate),
211        )));
212        self
213    }
214
215    /// Emit when the bucket has been inactive for the given duration.
216    pub fn complete_on_timeout(mut self, duration: Duration) -> Self {
217        self.completion = Some(CompletionMode::Single(CompletionCondition::Timeout(
218            duration,
219        )));
220        self
221    }
222
223    /// Emit when the bucket reaches `size` OR has been inactive for `timeout`.
224    pub fn complete_on_size_or_timeout(mut self, size: usize, timeout: Duration) -> Self {
225        self.completion = Some(CompletionMode::Any(vec![
226            CompletionCondition::Size(size),
227            CompletionCondition::Timeout(timeout),
228        ]));
229        self
230    }
231
232    /// Enable force-completion of pending buckets when the route is stopped.
233    pub fn force_completion_on_stop(mut self, enabled: bool) -> Self {
234        self.force_completion_on_stop = enabled;
235        self
236    }
237
238    /// Discard bucket contents on timeout instead of emitting the aggregated exchange.
239    pub fn discard_on_timeout(mut self, enabled: bool) -> Self {
240        self.discard_on_timeout = enabled;
241        self
242    }
243
244    /// Override the correlation strategy with a header-based key.
245    pub fn correlate_by(mut self, header: impl Into<String>) -> Self {
246        let header = header.into();
247        self.header_name = header.clone();
248        self.correlation = CorrelationStrategy::HeaderName(header);
249        self
250    }
251
252    /// Override the default `CollectAll` aggregation strategy.
253    pub fn strategy(mut self, strategy: AggregationStrategy) -> Self {
254        self.strategy = strategy;
255        self
256    }
257
258    /// Set the maximum number of correlation key buckets.
259    /// When the limit is reached, new correlation keys are rejected with an error.
260    pub fn max_buckets(mut self, max: usize) -> Self {
261        self.max_buckets = Some(max);
262        self
263    }
264
265    /// Set the time-to-live for inactive buckets.
266    /// Buckets that haven't been updated for this duration will be evicted.
267    pub fn bucket_ttl(mut self, ttl: Duration) -> Self {
268        self.bucket_ttl = Some(ttl);
269        self
270    }
271
272    /// Override the maximum number of concurrently-live per-bucket timeout tasks.
273    pub fn max_timeout_tasks(mut self, max: usize) -> Self {
274        self.max_timeout_tasks = max;
275        self
276    }
277
278    pub fn try_build(self) -> Result<AggregatorConfig, CamelError> {
279        // R3-C1 Batch 1: a completion-bound is mandatory. A config with no
280        // completion bound lets a bucket live forever — combined with a
281        // unique-key flood, that is the remote-OOM vector. Typed
282        // ConfigValidationError (ADR-0033) — operators can match on the
283        // `AggregatorMissingCompletionBound` variant.
284        let completion = self.completion.ok_or_else(|| {
285            CamelError::from(ConfigValidationError::AggregatorMissingCompletionBound)
286        })?;
287        Ok(AggregatorConfig {
288            header_name: self.header_name,
289            completion,
290            correlation: self.correlation,
291            strategy: self.strategy,
292            max_buckets: self.max_buckets,
293            bucket_ttl: self.bucket_ttl,
294            force_completion_on_stop: self.force_completion_on_stop,
295            discard_on_timeout: self.discard_on_timeout,
296            max_timeout_tasks: self.max_timeout_tasks,
297        })
298    }
299
300    /// Build the config. Returns an error if no completion condition was set.
301    pub fn build(self) -> Result<AggregatorConfig, CamelError> {
302        self.try_build()
303    }
304}
305
306#[cfg(test)]
307mod tests {
308    use super::*;
309
310    #[test]
311    fn test_aggregator_config_complete_when_size() {
312        let config = AggregatorConfig::correlate_by("orderId")
313            .complete_when_size(3)
314            .build()
315            .unwrap();
316        assert_eq!(config.header_name, "orderId");
317        assert!(matches!(
318            config.completion,
319            CompletionMode::Single(CompletionCondition::Size(3))
320        ));
321        assert!(matches!(config.strategy, AggregationStrategy::CollectAll));
322    }
323
324    #[test]
325    fn test_aggregator_config_complete_when_predicate() {
326        let config = AggregatorConfig::correlate_by("key")
327            .complete_when(|bucket| bucket.len() >= 2)
328            .build()
329            .unwrap();
330        assert!(matches!(
331            config.completion,
332            CompletionMode::Single(CompletionCondition::Predicate(_))
333        ));
334    }
335
336    #[test]
337    fn test_aggregator_config_custom_strategy() {
338        use std::sync::Arc;
339        let f: AggregationFn = Arc::new(|acc, _next| acc);
340        let config = AggregatorConfig::correlate_by("key")
341            .complete_when_size(1)
342            .strategy(AggregationStrategy::Custom(f))
343            .build()
344            .unwrap();
345        assert!(matches!(config.strategy, AggregationStrategy::Custom(_)));
346    }
347
348    #[test]
349    fn test_aggregator_config_missing_completion_returns_err() {
350        let result = AggregatorConfig::correlate_by("key").build();
351        let err = match result {
352            Err(e) => e,
353            Ok(_) => panic!("expected error, got Ok"),
354        };
355        assert!(
356            err.to_string().contains("completion"),
357            "error message should mention 'completion': {err}"
358        );
359    }
360
361    #[test]
362    fn test_complete_on_size_or_timeout() {
363        let config = AggregatorConfig::correlate_by("key")
364            .complete_on_size_or_timeout(3, Duration::from_secs(5))
365            .build()
366            .unwrap();
367        assert!(matches!(config.completion, CompletionMode::Any(v) if v.len() == 2));
368    }
369
370    #[test]
371    fn test_force_completion_on_stop_default() {
372        let config = AggregatorConfig::correlate_by("key")
373            .complete_when_size(1)
374            .build()
375            .unwrap();
376        assert!(!config.force_completion_on_stop);
377        assert!(!config.discard_on_timeout);
378    }
379
380    #[test]
381    fn test_builder_sets_timeout_and_flags_and_limits() {
382        let config = AggregatorConfig::correlate_by("key")
383            .complete_on_timeout(Duration::from_secs(2))
384            .max_buckets(7)
385            .bucket_ttl(Duration::from_secs(10))
386            .force_completion_on_stop(true)
387            .discard_on_timeout(true)
388            .build()
389            .unwrap();
390
391        assert!(matches!(
392            config.completion,
393            CompletionMode::Single(CompletionCondition::Timeout(d)) if d == Duration::from_secs(2)
394        ));
395        assert_eq!(config.max_buckets, Some(7));
396        assert_eq!(config.bucket_ttl, Some(Duration::from_secs(10)));
397        assert!(config.force_completion_on_stop);
398        assert!(config.discard_on_timeout);
399    }
400
401    #[test]
402    fn test_builder_correlate_by_overrides_header_and_strategy() {
403        let config = AggregatorConfig::correlate_by("original")
404            .correlate_by("override")
405            .complete_when_size(1)
406            .build()
407            .unwrap();
408
409        assert_eq!(config.header_name, "override");
410        assert!(matches!(
411            config.correlation,
412            CorrelationStrategy::HeaderName(ref h) if h == "override"
413        ));
414    }
415
416    #[test]
417    fn test_completion_reason_as_str_all_variants() {
418        assert_eq!(CompletionReason::Size.as_str(), "size");
419        assert_eq!(CompletionReason::Predicate.as_str(), "predicate");
420        assert_eq!(CompletionReason::Timeout.as_str(), "timeout");
421        assert_eq!(CompletionReason::Stop.as_str(), "stop");
422    }
423
424    #[test]
425    fn test_correlation_strategy_clone_and_debug() {
426        let strategy = CorrelationStrategy::Expression {
427            expr: "${header.orderId}".to_string(),
428            language: "simple".to_string(),
429        };
430        let cloned = strategy.clone();
431        assert!(matches!(
432            cloned,
433            CorrelationStrategy::Expression { ref expr, ref language }
434                if expr == "${header.orderId}" && language == "simple"
435        ));
436
437        let f = CorrelationStrategy::Fn(Arc::new(|_| Some("k".to_string())));
438        assert_eq!(format!("{:?}", f), "Fn(..)");
439    }
440
441    #[test]
442    fn test_complete_on_size_or_timeout_contains_both_conditions() {
443        let config = AggregatorConfig::correlate_by("k")
444            .complete_on_size_or_timeout(4, Duration::from_millis(250))
445            .build()
446            .unwrap();
447
448        match config.completion {
449            CompletionMode::Any(conditions) => {
450                assert!(matches!(conditions[0], CompletionCondition::Size(4)));
451                assert!(matches!(
452                    conditions[1],
453                    CompletionCondition::Timeout(d) if d == Duration::from_millis(250)
454                ));
455            }
456            _ => panic!("expected CompletionMode::Any"),
457        }
458    }
459
460    #[test]
461    #[allow(clippy::type_complexity)]
462    fn test_correlation_strategy_fn_clone_shares_same_arc() {
463        let f: Arc<dyn Fn(&Exchange) -> Option<String> + Send + Sync> =
464            Arc::new(|_| Some("shared".to_string()));
465        let strategy = CorrelationStrategy::Fn(f.clone());
466        let cloned = strategy.clone();
467
468        match cloned {
469            CorrelationStrategy::Fn(cloned_fn) => assert!(Arc::ptr_eq(&f, &cloned_fn)),
470            _ => panic!("expected fn strategy"),
471        }
472    }
473
474    #[test]
475    fn test_builder_correlate_by_overrides_previous() {
476        let config = AggregatorConfig::correlate_by("first")
477            .correlate_by("second")
478            .complete_when_size(2)
479            .build()
480            .unwrap();
481
482        assert_eq!(config.header_name, "second");
483        assert!(
484            matches!(config.correlation, CorrelationStrategy::HeaderName(ref h) if h == "second")
485        );
486    }
487
488    #[test]
489    fn test_aggregator_try_build_missing_completion_returns_error() {
490        let result = AggregatorConfig::correlate_by("key").try_build();
491        assert!(result.is_err());
492    }
493
494    // ── R3-C1 Batch 1: DoS caps + completion-bound validation ────────
495
496    /// The builder default for `max_buckets` is `Some(10_000)`. The spec fixes
497    /// this as the bounded default; the operator may still override.
498    #[test]
499    fn test_default_max_buckets_is_10000() {
500        let cfg = AggregatorConfig::correlate_by("k")
501            .complete_when_size(1)
502            .build()
503            .unwrap();
504        assert_eq!(cfg.max_buckets, Some(10_000));
505    }
506
507    /// The builder default for `bucket_ttl` is `Some(Duration::from_secs(300))`.
508    /// Inline retain + background sweep both use this TTL.
509    #[test]
510    fn test_default_bucket_ttl_is_300s() {
511        let cfg = AggregatorConfig::correlate_by("k")
512            .complete_when_size(1)
513            .build()
514            .unwrap();
515        assert_eq!(cfg.bucket_ttl, Some(Duration::from_secs(300)));
516    }
517
518    /// Configs that override `max_buckets(0)` still build (the operator chose it).
519    /// The cap may be set to 1 by the operator; the bound itself is not validated.
520    /// The completion-bound check below is what Batch 1 enforces.
521    #[test]
522    fn test_explicit_max_buckets_zero_is_accepted_at_build() {
523        let cfg = AggregatorConfig::correlate_by("k")
524            .complete_when_size(1)
525            .max_buckets(0)
526            .build()
527            .unwrap();
528        assert_eq!(cfg.max_buckets, Some(0));
529    }
530
531    /// A config with no completion bound — neither size, nor timeout, nor predicate —
532    /// is rejected at `try_build` with `AggregatorMissingCompletionBound`. Spec §11
533    /// RESOLVED: at least one completion bound is mandatory.
534    #[test]
535    fn test_aggregator_rejects_no_completion_bound() {
536        // Builder has no `complete_*` call → try_build returns Err.
537        // Use match (not unwrap_err) because AggregatorConfig is not Debug.
538        let err = match AggregatorConfig::correlate_by("k").try_build() {
539            Err(e) => e,
540            Ok(_) => panic!("expected error, got Ok"),
541        };
542        assert!(
543            matches!(
544                err,
545                CamelError::ConfigValidation(
546                    ConfigValidationError::AggregatorMissingCompletionBound
547                )
548            ),
549            "expected ConfigValidation(AggregatorMissingCompletionBound), got: {err}"
550        );
551    }
552
553    // ── R3-M2: memory-bound validation ────────────────────────────────
554
555    #[test]
556    fn test_aggregator_config_rejects_no_memory_bound() {
557        // Direct construction bypassing the builder defaults — simulate a config
558        // with no max_buckets, no timeout, no ttl.
559        let config = AggregatorConfig {
560            header_name: "k".into(),
561            completion: CompletionMode::Single(CompletionCondition::Size(2)),
562            correlation: CorrelationStrategy::HeaderName("k".into()),
563            strategy: AggregationStrategy::CollectAll,
564            max_buckets: None,
565            bucket_ttl: None,
566            force_completion_on_stop: false,
567            discard_on_timeout: false,
568            max_timeout_tasks: 1024,
569        };
570        let err = config.validate().unwrap_err();
571        assert!(
572            err.to_string().contains("max_buckets")
573                || err.to_string().contains("completionTimeout")
574                || err.to_string().contains("bucket_ttl"),
575            "error should explain the required bound: {err}"
576        );
577    }
578
579    #[test]
580    fn test_aggregator_config_accepts_size_only_with_max_buckets() {
581        // Builder path defaults max_buckets + bucket_ttl — must validate OK.
582        let config = AggregatorConfig::correlate_by("k")
583            .complete_when_size(2)
584            .build()
585            .unwrap();
586        assert!(config.validate().is_ok());
587    }
588
589    /// R3-M3: Timeout completion requires bucket_ttl so the cap-skip
590    /// degradation always has an eviction path. A config with Timeout but
591    /// no bucket_ttl is rejected by validate().
592    #[test]
593    fn test_aggregator_timeout_requires_bucket_ttl() {
594        let config = AggregatorConfig {
595            header_name: "k".into(),
596            completion: CompletionMode::Single(CompletionCondition::Timeout(Duration::from_secs(
597                5,
598            ))),
599            correlation: CorrelationStrategy::HeaderName("k".into()),
600            strategy: AggregationStrategy::CollectAll,
601            max_buckets: Some(100),
602            bucket_ttl: None, // <-- missing ttl fallback
603            force_completion_on_stop: false,
604            discard_on_timeout: false,
605            max_timeout_tasks: 1024,
606        };
607        let err = config.validate().unwrap_err();
608        assert!(
609            err.to_string().contains("bucket_ttl") || err.to_string().contains("Timeout"),
610            "error should explain the timeout-requires-ttl invariant: {err}"
611        );
612    }
613}
camel_api/aggregator.rs

camel_api/
aggregator.rs
