camel_processor/
cache_eip.rs

1//! Cache EIP — outcome-aware Segment implementation.
2//!
3//! Implements the Caching pattern (lookup → on-miss sub-pipeline → write-back)
4//! at the `OutcomePipeline` layer (one layer above Tower), mirroring
5//! [`IdempotentConsumerSegment`]. On a cache HIT the body is reconstructed from
6//! the stored [`CacheEntry`] and the on-miss sub-pipeline is skipped entirely;
7//! on a MISS the sub-pipeline runs and its result body is written back into the
8//! repository (subject to `max_entry_bytes`).
9//!
10//! # Why Segment-mode (NOT Process-mode)
11//!
12//! Same rationale as the idempotent consumer: a Tower `Service<Exchange>` cannot
13//! propagate `PipelineOutcome::Stopped` distinctly from `Ok(ex)`. By implementing
14//! [`OutcomePipeline`] directly, a `Stopped` from the on-miss sub-pipeline flows
15//! out with the Exchange intact and NO write-back occurs (ADR-0024, ADR-0025).
16//!
17//! # Contract C1 (ADR-0023)
18//!
19//! [`CacheRepository::get`] / [`CacheRepository::set`] surface backend failures
20//! as `Err(CamelError)`. The segment propagates those as `PipelineOutcome::Failed`
21//! — it NEVER treats a failed read as a miss.
22
23use std::future::Future;
24use std::pin::Pin;
25use std::sync::Arc;
26use std::time::{Duration, SystemTime};
27
28use bytes::Bytes;
29
30use camel_api::body::Body;
31use camel_api::cache::{CacheEntry, CacheRepository, ContentType};
32use camel_api::{CamelError, Exchange, OutcomePipeline, OutcomeSegment, PipelineOutcome};
33use camel_component_api::RuntimeObservability;
34
35use crate::MessageIdExpression;
36
37/// Outcome-aware Cache segment (Caching EIP).
38///
39/// Wraps a named [`CacheRepository`] and an on-miss sub-pipeline
40/// ([`OutcomeSegment`]). On each exchange:
41///
42/// 1. Evaluate `key_expr`. `None` → not cacheable; forward directly to the
43///    on-miss sub-pipeline (no lookup, no write-back).
44/// 2. `repository.get(&key)`:
45///    - `Err(e)` → `Failed(e)` (contract C1).
46///    - `Ok(Some(entry))` → HIT: reconstruct `Body` from the entry, set it on
47///      the exchange, return `Completed` (skip on-miss).
48///    - `Ok(None)` → MISS: proceed to step 3.
49/// 3. Run the on-miss sub-pipeline.
50///    - `Stopped(ex)` / `Failed(e)` → propagate as-is (NO write-back).
51///    - `Completed(ex)` → proceed to write-back.
52/// 4. Write-back the resulting body (when it fits `max_entry_bytes`):
53///    - materialized variants (`Bytes`/`Text`/`Json`/`Xml`) → serialize, store.
54///    - `Stream` → materialize via [`Body::into_bytes`] (consumes the body,
55///      replaces it with `Body::Bytes`); `StreamLimitExceeded` propagates.
56///    - `Empty` / oversized body → pass through uncached, return `Completed`.
57pub struct CacheService {
58    repository: Arc<dyn CacheRepository>,
59    /// Cached `repository.name()` for OTel span tagging (Task 3.3).
60    repository_name: String,
61    key_expr: MessageIdExpression,
62    ttl: Option<Duration>,
63    max_entry_bytes: usize,
64    on_miss: OutcomeSegment,
65    rt: Arc<dyn RuntimeObservability>,
66}
67
68impl CacheService {
69    /// Build a new cache segment.
70    ///
71    /// `repository_name` is derived from `repository.name()` so OTel tags stay
72    /// in sync with the resolved backend.
73    pub fn new(
74        repository: Arc<dyn CacheRepository>,
75        key_expr: MessageIdExpression,
76        ttl: Option<Duration>,
77        max_entry_bytes: usize,
78        on_miss: OutcomeSegment,
79        rt: Arc<dyn RuntimeObservability>,
80    ) -> Self {
81        let repository_name = repository.name().to_string();
82        Self {
83            repository,
84            repository_name,
85            key_expr,
86            ttl,
87            max_entry_bytes,
88            on_miss,
89            rt,
90        }
91    }
92
93    /// The configured repository name (for OTel tagging).
94    pub fn repository_name(&self) -> &str {
95        &self.repository_name
96    }
97}
98
99/// Shared write-back tail for materialized bodies.
100///
101/// Checks `max_entry_bytes`, builds a [`CacheEntry`], stores via
102/// the repository, and returns `Completed(exchange)`. The exchange body
103/// is not modified — it passes through as-is. On oversized body, logs a
104/// debug! skip message and returns `Completed(exchange)` without storing.
105/// On repository error, returns `Failed(e)`.
106#[allow(clippy::too_many_arguments)]
107async fn write_back(
108    repository: &Arc<dyn CacheRepository>,
109    repository_name: &str,
110    max_entry_bytes: usize,
111    ttl: Option<Duration>,
112    exchange: Exchange,
113    key: &str,
114    serialized: Vec<u8>,
115    content_type: ContentType,
116) -> PipelineOutcome {
117    if serialized.len() <= max_entry_bytes {
118        let entry = CacheEntry {
119            bytes: serialized,
120            content_type,
121            expires_at: None,
122        };
123        match repository.set(key, entry, ttl).await {
124            Ok(()) => {}
125            Err(e) => {
126                if matches!(&e, CamelError::Config(msg) if msg.starts_with("cache: max_entries")) {
127                    tracing::debug!(
128                        repository = %repository_name,
129                        key = %key,
130                        "cache at capacity, skipping write-back"
131                    ); // log-policy: g:cache:capacity-full-skip
132                } else {
133                    return PipelineOutcome::Failed(e);
134                }
135            }
136        }
137    } else {
138        // log-policy: g:cache:oversized-skip
139        tracing::debug!(
140            repository = %repository_name,
141            key = %key,
142            len = serialized.len(),
143            max = max_entry_bytes,
144            "cache write-back skipped: body exceeds max_entry_bytes"
145        );
146    }
147    PipelineOutcome::Completed(exchange)
148}
149
150impl Clone for CacheService {
151    fn clone(&self) -> Self {
152        Self {
153            repository: Arc::clone(&self.repository),
154            repository_name: self.repository_name.clone(),
155            key_expr: Arc::clone(&self.key_expr),
156            ttl: self.ttl,
157            max_entry_bytes: self.max_entry_bytes,
158            on_miss: self.on_miss.clone(),
159            rt: Arc::clone(&self.rt),
160        }
161    }
162}
163
164impl OutcomePipeline for CacheService {
165    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
166        Box::new(self.clone())
167    }
168
169    fn run<'a>(
170        &'a mut self,
171        exchange: Exchange,
172    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
173        Box::pin(async move {
174            // 1. Evaluate key. None → not cacheable, bypass straight to on_miss.
175            let key = match (self.key_expr)(&exchange) {
176                Some(k) => k,
177                None => return self.on_miss.run(exchange).await,
178            };
179
180            // 2. Lookup (contract C1: propagate Err, never treat as miss).
181            match self.repository.get(&key).await {
182                Err(e) => return PipelineOutcome::Failed(e),
183                Ok(Some(entry)) => {
184                    // HIT: record metric, reconstruct body, skip on-miss sub-pipeline.
185                    self.rt.metrics().record_counter(
186                        "camel.cache.hits",
187                        1.0_f64,
188                        &[("repository", &self.repository_name)],
189                    );
190                    match reconstruct_body(&entry) {
191                        Ok(body) => {
192                            let mut exchange = exchange;
193                            exchange.input.body = body;
194                            return PipelineOutcome::Completed(exchange);
195                        }
196                        Err(e) => return PipelineOutcome::Failed(e),
197                    }
198                }
199                Ok(None) => {
200                    // MISS: record metric, fall through to on-miss sub-pipeline.
201                    self.rt.metrics().record_counter(
202                        "camel.cache.misses",
203                        1.0_f64,
204                        &[("repository", &self.repository_name)],
205                    );
206                }
207            }
208
209            // 3. Run the on-miss sub-pipeline.
210            let mut exchange = match self.on_miss.run(exchange).await {
211                PipelineOutcome::Stopped(ex) => return PipelineOutcome::Stopped(ex),
212                PipelineOutcome::Failed(e) => return PipelineOutcome::Failed(e),
213                PipelineOutcome::Completed(ex) => ex,
214            };
215
216            // 4. Write-back. Take the body out so the Stream arm can consume it.
217            let body = std::mem::replace(&mut exchange.input.body, Body::Empty);
218            match body {
219                Body::Bytes(b) => {
220                    let serialized = b.to_vec();
221                    exchange.input.body = Body::Bytes(b);
222                    write_back(
223                        &self.repository,
224                        &self.repository_name,
225                        self.max_entry_bytes,
226                        self.ttl,
227                        exchange,
228                        &key,
229                        serialized,
230                        ContentType::Bytes,
231                    )
232                    .await
233                }
234                Body::Text(s) => {
235                    let serialized = s.as_bytes().to_vec();
236                    exchange.input.body = Body::Text(s);
237                    write_back(
238                        &self.repository,
239                        &self.repository_name,
240                        self.max_entry_bytes,
241                        self.ttl,
242                        exchange,
243                        &key,
244                        serialized,
245                        ContentType::Text,
246                    )
247                    .await
248                }
249                Body::Json(v) => {
250                    let serialized = match serde_json::to_vec(&v) {
251                        Ok(b) => b,
252                        Err(e) => {
253                            exchange.input.body = Body::Json(v);
254                            return PipelineOutcome::Failed(CamelError::TypeConversionFailed(
255                                e.to_string(),
256                            ));
257                        }
258                    };
259                    exchange.input.body = Body::Json(v);
260                    write_back(
261                        &self.repository,
262                        &self.repository_name,
263                        self.max_entry_bytes,
264                        self.ttl,
265                        exchange,
266                        &key,
267                        serialized,
268                        ContentType::Json,
269                    )
270                    .await
271                }
272                Body::Xml(s) => {
273                    let serialized = s.as_bytes().to_vec();
274                    exchange.input.body = Body::Xml(s);
275                    write_back(
276                        &self.repository,
277                        &self.repository_name,
278                        self.max_entry_bytes,
279                        self.ttl,
280                        exchange,
281                        &key,
282                        serialized,
283                        ContentType::Xml,
284                    )
285                    .await
286                }
287                Body::Stream(stream_body) => {
288                    // Materialize (consumes the stream). StreamLimitExceeded propagates.
289                    let materialized = match Body::Stream(stream_body)
290                        .into_bytes(self.max_entry_bytes)
291                        .await
292                    {
293                        Ok(b) => b,
294                        Err(e) => return PipelineOutcome::Failed(e),
295                    };
296                    // into_bytes already enforced max_entry_bytes, so it fits by construction.
297                    let entry = CacheEntry {
298                        bytes: materialized.to_vec(),
299                        content_type: ContentType::Bytes,
300                        expires_at: None,
301                    };
302                    if let Err(e) = self.repository.set(&key, entry, self.ttl).await {
303                        // Degrade capacity-exceeded to uncached — same policy as write_back.
304                        if matches!(&e, CamelError::Config(msg) if msg.starts_with("cache: max_entries"))
305                        {
306                            tracing::debug!(
307                                repository = %self.repository_name,
308                                key = %key,
309                                "cache at capacity, skipping write-back for stream"
310                            ); // log-policy: g:cache:capacity-full-skip
311                            exchange.input.body = Body::Bytes(materialized);
312                            return PipelineOutcome::Completed(exchange);
313                        }
314                        exchange.input.body = Body::Bytes(materialized);
315                        return PipelineOutcome::Failed(e);
316                    }
317                    exchange.input.body = Body::Bytes(materialized);
318                    PipelineOutcome::Completed(exchange)
319                }
320                _ => {
321                    // Empty (or any future variant): pass through uncached.
322                    exchange.input.body = body;
323                    PipelineOutcome::Completed(exchange)
324                }
325            }
326        })
327    }
328}
329
330/// Reconstruct a [`Body`] from a stored [`CacheEntry`].
331///
332/// Maps each [`ContentType`] back to the matching `Body` variant, decoding
333/// UTF-8 / JSON failures into `CamelError::TypeConversionFailed`.
334fn reconstruct_body(entry: &CacheEntry) -> Result<Body, CamelError> {
335    match entry.content_type {
336        ContentType::Bytes => Ok(Body::Bytes(Bytes::from(entry.bytes.clone()))),
337        ContentType::Text => {
338            let s = String::from_utf8(entry.bytes.clone()).map_err(|e| {
339                CamelError::TypeConversionFailed(format!("cached text is not valid UTF-8: {e}"))
340            })?;
341            Ok(Body::Text(s))
342        }
343        ContentType::Json => {
344            let v = serde_json::from_slice(&entry.bytes).map_err(|e| {
345                CamelError::TypeConversionFailed(format!("cached bytes are not valid JSON: {e}"))
346            })?;
347            Ok(Body::Json(v))
348        }
349        ContentType::Xml => {
350            let s = String::from_utf8(entry.bytes.clone()).map_err(|e| {
351                CamelError::TypeConversionFailed(format!("cached xml is not valid UTF-8: {e}"))
352            })?;
353            Ok(Body::Xml(s))
354        }
355    }
356}
357
358// ===========================================================================
359// CacheInvalidateService — invalidate a single cache entry
360// ===========================================================================
361
362/// Outcome-aware segment that invalidates a single cache entry.
363///
364/// Evaluates `key_expr`:
365/// - `None` → `Completed(exchange)` (nothing to invalidate).
366/// - `Some(key)` → `repository.invalidate(&key).await`.
367///   - `Err(e)` → `Failed(e)`.
368///   - `Ok(())` → `Completed(exchange)`.
369pub struct CacheInvalidateService {
370    repository: Arc<dyn CacheRepository>,
371    key_expr: MessageIdExpression,
372}
373
374impl CacheInvalidateService {
375    pub fn new(repository: Arc<dyn CacheRepository>, key_expr: MessageIdExpression) -> Self {
376        Self {
377            repository,
378            key_expr,
379        }
380    }
381}
382
383impl Clone for CacheInvalidateService {
384    fn clone(&self) -> Self {
385        Self {
386            repository: Arc::clone(&self.repository),
387            key_expr: Arc::clone(&self.key_expr),
388        }
389    }
390}
391
392impl OutcomePipeline for CacheInvalidateService {
393    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
394        Box::new(self.clone())
395    }
396
397    fn run<'a>(
398        &'a mut self,
399        exchange: Exchange,
400    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
401        Box::pin(async move {
402            let key = match (self.key_expr)(&exchange) {
403                Some(k) => k,
404                None => return PipelineOutcome::Completed(exchange),
405            };
406            match self.repository.invalidate(&key).await {
407                Err(e) => PipelineOutcome::Failed(e),
408                Ok(()) => PipelineOutcome::Completed(exchange),
409            }
410        })
411    }
412}
413
414// ===========================================================================
415// CachePeekStaleService — serve a stale entry after expiry
416// ===========================================================================
417
418/// Exchange property set to `true` when a `cache_peek_stale` HIT occurred.
419pub const CAMEL_CACHE_PEEK_HIT: &str = "CamelCachePeekHit";
420/// Exchange property set to `true` when the served entry was stale (post-expiry).
421pub const CAMEL_CACHE_PEEK_STALE: &str = "CamelCachePeekStale";
422
423/// On-miss policy for [`CachePeekStaleService`].
424///
425/// - [`Stop`](PeekStaleMissPolicy::Stop) (default) preserves the
426///   `CircuitBreaker.fallback` absence-Stops contract.
427/// - [`Continue`](PeekStaleMissPolicy::Continue) leaves the body untouched on
428///   MISS so `choice` can branch on [`CAMEL_CACHE_PEEK_HIT`].
429#[derive(Debug, Clone, Copy, PartialEq, Eq)]
430pub enum PeekStaleMissPolicy {
431    /// MISS Stops the branch (no stale available — `CircuitBreaker.fallback`).
432    Stop,
433    /// MISS continues with the body unchanged.
434    Continue,
435}
436
437impl PeekStaleMissPolicy {
438    /// Parses the canonical/DSL `cache_peek_stale.on_miss` knob:
439    /// absent or `"stop"` → [`Stop`](Self::Stop), `"continue"` →
440    /// [`Continue`](Self::Continue). Any other value fails closed naming
441    /// the step.
442    pub fn parse_on_miss(raw: Option<&str>) -> Result<Self, CamelError> {
443        match raw {
444            None | Some("stop") => Ok(Self::Stop),
445            Some("continue") => Ok(Self::Continue),
446            Some(other) => Err(CamelError::Config(format!(
447                "cache_peek_stale: invalid on_miss '{other}'; must be \"stop\" or \"continue\""
448            ))),
449        }
450    }
451}
452
453/// Outcome-aware segment that serves a stale (post-expiry) cache entry.
454///
455/// Evaluates `key_expr`:
456/// - `None` → `Stopped(exchange)` with a `debug` log (an anomalous key
457///   resolution is fail-closed, not a miss).
458/// - `Some(key)` → `repository.peek_stale(&key).await`.
459///   - `Err(e)` → `Failed(e)`.
460///   - `Ok(Some(entry))` → reconstruct body from entry, set
461///     `CamelCachePeekHit=true` and `CamelCachePeekStale` (true when the
462///     entry's `expires_at` has elapsed at evaluation time; false when absent
463///     or not elapsed), return `Completed(exchange)`.
464///   - `Ok(None)` → MISS (absence), governed by [`PeekStaleMissPolicy`]:
465///     - `Stop` (default): set `CamelCachePeekHit=false` and
466///       `CamelCachePeekStale=false`, log at `debug`, return `Stopped(exchange)`
467///       (absence in `CircuitBreaker.fallback` means "no stale available").
468///     - `Continue`: set `CamelCachePeekHit=false` and
469///       `CamelCachePeekStale=false`, leave the body unchanged, return
470///       `Completed(exchange)` so `choice` can branch on `CamelCachePeekHit`.
471pub struct CachePeekStaleService {
472    repository: Arc<dyn CacheRepository>,
473    key_expr: MessageIdExpression,
474    miss_policy: PeekStaleMissPolicy,
475}
476
477impl CachePeekStaleService {
478    pub fn new(
479        repository: Arc<dyn CacheRepository>,
480        key_expr: MessageIdExpression,
481        miss_policy: PeekStaleMissPolicy,
482    ) -> Self {
483        Self {
484            repository,
485            key_expr,
486            miss_policy,
487        }
488    }
489}
490
491impl Clone for CachePeekStaleService {
492    fn clone(&self) -> Self {
493        Self {
494            repository: Arc::clone(&self.repository),
495            key_expr: Arc::clone(&self.key_expr),
496            miss_policy: self.miss_policy,
497        }
498    }
499}
500
501/// Write the peek-result exchange properties under [`CAMEL_CACHE_PEEK_HIT`] and
502/// [`CAMEL_CACHE_PEEK_STALE`] as `serde_json::Value::Bool` values.
503fn set_peek_properties(exchange: &mut Exchange, hit: bool, stale: bool) {
504    exchange.set_property(CAMEL_CACHE_PEEK_HIT, serde_json::Value::Bool(hit));
505    exchange.set_property(CAMEL_CACHE_PEEK_STALE, serde_json::Value::Bool(stale));
506}
507
508impl OutcomePipeline for CachePeekStaleService {
509    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
510        Box::new(self.clone())
511    }
512
513    fn run<'a>(
514        &'a mut self,
515        exchange: Exchange,
516    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
517        Box::pin(async move {
518            let key = match (self.key_expr)(&exchange) {
519                Some(k) => k,
520                None => {
521                    tracing::debug!(
522                        step = "cache_peek_stale",
523                        repository = %self.repository.name(),
524                        "key expression resolved to None; stopping branch"
525                    );
526                    return PipelineOutcome::Stopped(exchange);
527                }
528            };
529            match self.repository.peek_stale(&key).await {
530                Err(e) => PipelineOutcome::Failed(e),
531                Ok(Some(entry)) => {
532                    // Staleness read before body reconstruction; reconstruct_body borrows the entry, so ordering is stylistic.
533                    let stale = entry
534                        .expires_at
535                        .map(|t| t <= SystemTime::now())
536                        .unwrap_or(false);
537                    match reconstruct_body(&entry) {
538                        Ok(body) => {
539                            let mut exchange = exchange;
540                            exchange.input.body = body;
541                            set_peek_properties(&mut exchange, true, stale);
542                            PipelineOutcome::Completed(exchange)
543                        }
544                        Err(e) => PipelineOutcome::Failed(e),
545                    }
546                }
547                Ok(None) => match self.miss_policy {
548                    PeekStaleMissPolicy::Stop => {
549                        let mut exchange = exchange;
550                        set_peek_properties(&mut exchange, false, false);
551                        tracing::debug!(
552                            step = "cache_peek_stale",
553                            repository = %self.repository.name(),
554                            "peek miss; stopping branch per on_miss=stop"
555                        );
556                        PipelineOutcome::Stopped(exchange)
557                    }
558                    PeekStaleMissPolicy::Continue => {
559                        let mut exchange = exchange;
560                        set_peek_properties(&mut exchange, false, false);
561                        PipelineOutcome::Completed(exchange)
562                    }
563                },
564            }
565        })
566    }
567}
568
569// ===========================================================================
570// Test utilities
571// ===========================================================================
572
573#[cfg(test)]
574mod test_utils {
575    use super::*;
576    use async_trait::async_trait;
577    use std::collections::HashMap;
578    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
579    use tokio::sync::Mutex;
580
581    /// In-memory mock [`CacheRepository`] for cache segment tests. Allows tests
582    /// to pre-seed entries, force `get`/`set` failures, and inspect the last
583    /// `set` call (entry + TTL).
584    #[derive(Debug, Default)]
585    pub struct MockCacheRepository {
586        name: String,
587        entries: Arc<Mutex<HashMap<String, CacheEntry>>>,
588        get_should_fail: Arc<AtomicBool>,
589        set_should_fail: Arc<AtomicBool>,
590        set_call_count: Arc<AtomicU32>,
591        last_set_ttl: Arc<Mutex<Option<Duration>>>,
592        invalidate_call_count: Arc<AtomicU32>,
593        last_invalidate_key: Arc<Mutex<Option<String>>>,
594    }
595
596    impl MockCacheRepository {
597        pub fn new(name: &str) -> Self {
598            Self {
599                name: name.to_string(),
600                ..Default::default()
601            }
602        }
603
604        pub fn invalidate_call_count(&self) -> u32 {
605            self.invalidate_call_count.load(Ordering::SeqCst)
606        }
607
608        pub async fn last_invalidate_key(&self) -> Option<String> {
609            self.last_invalidate_key.lock().await.clone()
610        }
611
612        /// Pre-seed a key so `get` returns a HIT.
613        pub async fn seed(&self, key: &str, entry: CacheEntry) {
614            self.entries.lock().await.insert(key.to_string(), entry);
615        }
616
617        pub fn set_get_should_fail(&self, v: bool) {
618            self.get_should_fail.store(v, Ordering::SeqCst);
619        }
620
621        pub fn set_set_should_fail(&self, v: bool) {
622            self.set_should_fail.store(v, Ordering::SeqCst);
623        }
624
625        pub fn set_call_count(&self) -> u32 {
626            self.set_call_count.load(Ordering::SeqCst)
627        }
628
629        /// The TTL passed to the most recent `set` call.
630        pub async fn last_set_ttl(&self) -> Option<Duration> {
631            *self.last_set_ttl.lock().await
632        }
633
634        /// Inspect the entry currently stored for `key` (if any).
635        pub async fn stored_entry(&self, key: &str) -> Option<CacheEntry> {
636            self.entries.lock().await.get(key).cloned()
637        }
638    }
639
640    #[async_trait]
641    impl CacheRepository for MockCacheRepository {
642        fn name(&self) -> &str {
643            &self.name
644        }
645
646        async fn get(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
647            if self.get_should_fail.load(Ordering::SeqCst) {
648                return Err(CamelError::ProcessorError("synthetic get failure".into()));
649            }
650            Ok(self.entries.lock().await.get(key).cloned())
651        }
652
653        async fn set(
654            &self,
655            key: &str,
656            value: CacheEntry,
657            ttl: Option<Duration>,
658        ) -> Result<(), CamelError> {
659            self.set_call_count.fetch_add(1, Ordering::SeqCst);
660            *self.last_set_ttl.lock().await = ttl;
661            if self.set_should_fail.load(Ordering::SeqCst) {
662                return Err(CamelError::ProcessorError("synthetic set failure".into()));
663            }
664            self.entries.lock().await.insert(key.to_string(), value);
665            Ok(())
666        }
667
668        async fn peek_stale(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
669            self.get(key).await
670        }
671
672        async fn invalidate(&self, key: &str) -> Result<(), CamelError> {
673            self.invalidate_call_count.fetch_add(1, Ordering::SeqCst);
674            *self.last_invalidate_key.lock().await = Some(key.to_string());
675            self.entries.lock().await.remove(key);
676            Ok(())
677        }
678
679        async fn clear(&self) -> Result<(), CamelError> {
680            self.entries.lock().await.clear();
681            Ok(())
682        }
683    }
684}
685
686// ===========================================================================
687// Tests
688// ===========================================================================
689
690#[cfg(test)]
691mod tests {
692    use super::test_utils::MockCacheRepository;
693    use super::*;
694    use camel_api::body::{StreamBody, StreamMetadata};
695    use camel_api::metrics::NoOpMetrics;
696    use camel_api::{Message, Value};
697    use camel_component_api::health_registry::NoOpHealthCheckRegistry;
698    use futures::stream;
699    use std::sync::Mutex;
700    use std::sync::atomic::{AtomicBool, Ordering};
701    use std::time::SystemTime;
702
703    #[test]
704    fn parse_on_miss_maps_absent_stop_and_continue() {
705        assert_eq!(
706            PeekStaleMissPolicy::parse_on_miss(None).unwrap(),
707            PeekStaleMissPolicy::Stop
708        );
709        assert_eq!(
710            PeekStaleMissPolicy::parse_on_miss(Some("stop")).unwrap(),
711            PeekStaleMissPolicy::Stop
712        );
713        assert_eq!(
714            PeekStaleMissPolicy::parse_on_miss(Some("continue")).unwrap(),
715            PeekStaleMissPolicy::Continue
716        );
717    }
718
719    #[test]
720    fn parse_on_miss_rejects_unknown_value_naming_the_step() {
721        let err = PeekStaleMissPolicy::parse_on_miss(Some("explode")).unwrap_err();
722        let msg = format!("{err}");
723        assert!(msg.contains("cache_peek_stale"), "got: {msg}");
724        assert!(msg.contains("explode"), "got: {msg}");
725    }
726
727    /// Minimal no-op RuntimeObservability for tests that don't need OTel.
728    #[derive(Clone)]
729    struct NoopRt;
730
731    impl camel_component_api::HealthCheckRegistry for NoopRt {
732        fn force_unhealthy_for_route(&self, _: &str, _: &str, _: &str) {}
733    }
734
735    impl RuntimeObservability for NoopRt {
736        fn metrics(&self) -> Arc<dyn camel_api::metrics::MetricsCollector> {
737            Arc::new(NoOpMetrics)
738        }
739        fn health(&self) -> Arc<dyn camel_component_api::health_registry::HealthCheckRegistry> {
740            Arc::new(NoOpHealthCheckRegistry)
741        }
742    }
743
744    fn noop_rt() -> Arc<dyn RuntimeObservability> {
745        Arc::new(NoopRt)
746    }
747
748    // ── Scripted on-miss sub-pipeline ──
749
750    #[derive(Clone)]
751    enum ScriptedOutcome {
752        Complete,
753        Stop,
754        Fail(CamelError),
755    }
756
757    /// Test sub-pipeline: optionally replaces the body, then returns a
758    /// scripted outcome. Records whether it was invoked.
759    struct ScriptedOnMiss {
760        body: Option<Body>,
761        outcome: ScriptedOutcome,
762        invoked: Arc<AtomicBool>,
763    }
764
765    impl OutcomePipeline for ScriptedOnMiss {
766        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
767            // clone_box is required by the trait but unused by these tests.
768            unreachable!("clone_box not used in cache_eip tests")
769        }
770
771        fn run<'a>(
772            &'a mut self,
773            mut exchange: Exchange,
774        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
775            self.invoked.store(true, Ordering::SeqCst);
776            let body = self.body.take();
777            let outcome = self.outcome.clone();
778            Box::pin(async move {
779                if let Some(b) = body {
780                    exchange.input.body = b;
781                }
782                match outcome {
783                    ScriptedOutcome::Complete => PipelineOutcome::Completed(exchange),
784                    ScriptedOutcome::Stop => PipelineOutcome::Stopped(exchange),
785                    ScriptedOutcome::Fail(e) => PipelineOutcome::Failed(e),
786                }
787            })
788        }
789    }
790
791    // ── Builders ──
792
793    fn fixed_key() -> MessageIdExpression {
794        Arc::new(|_| Some("cache-key".to_string()))
795    }
796
797    fn none_key() -> MessageIdExpression {
798        Arc::new(|_| None)
799    }
800
801    /// Build a CacheService whose on-miss sets `body` and returns `outcome`.
802    fn build_service(
803        repo: Arc<MockCacheRepository>,
804        key_expr: MessageIdExpression,
805        max_entry_bytes: usize,
806        body: Option<Body>,
807        outcome: ScriptedOutcome,
808        ttl: Option<Duration>,
809        rt: Arc<dyn RuntimeObservability>,
810    ) -> (CacheService, Arc<AtomicBool>) {
811        let invoked = Arc::new(AtomicBool::new(false));
812        let on_miss = OutcomeSegment::new(Box::new(ScriptedOnMiss {
813            body,
814            outcome,
815            invoked: invoked.clone(),
816        }));
817        let svc = CacheService::new(repo, key_expr, ttl, max_entry_bytes, on_miss, rt);
818        (svc, invoked)
819    }
820
821    fn exchange() -> Exchange {
822        let mut ex = Exchange::new(Message::new(""));
823        ex.input.set_header("ignored", Value::String("v".into()));
824        ex
825    }
826
827    fn stream_body(data: &'static [u8]) -> Body {
828        let chunks: Vec<Result<Bytes, CamelError>> = vec![Ok(Bytes::from_static(data))];
829        let s = stream::iter(chunks);
830        Body::Stream(StreamBody {
831            stream: Arc::new(tokio::sync::Mutex::new(Some(Box::pin(s)))),
832            metadata: StreamMetadata::default(),
833        })
834    }
835
836    fn stub_error(msg: &str) -> CamelError {
837        CamelError::ProcessorError(msg.into())
838    }
839
840    // ── Test 1: cache HIT short-circuits, on_miss NOT executed ──
841
842    #[tokio::test]
843    async fn cache_hit_short_circuits_on_miss() {
844        let repo = Arc::new(MockCacheRepository::new("mock"));
845        repo.seed(
846            "cache-key",
847            CacheEntry {
848                bytes: b"cached-payload".to_vec(),
849                content_type: ContentType::Bytes,
850                expires_at: None,
851            },
852        )
853        .await;
854        let (mut svc, on_miss_invoked) = build_service(
855            repo,
856            fixed_key(),
857            1024,
858            Some(Body::Bytes(Bytes::from_static(b"unreached"))),
859            ScriptedOutcome::Complete,
860            None,
861            noop_rt(),
862        );
863
864        let outcome = svc.run(exchange()).await;
865
866        let ex = match outcome {
867            PipelineOutcome::Completed(ex) => ex,
868            other => panic!("expected Completed, got {other:?}"),
869        };
870        assert_eq!(
871            ex.input.body,
872            Body::Bytes(Bytes::from_static(b"cached-payload"))
873        );
874        assert!(
875            !on_miss_invoked.load(Ordering::SeqCst),
876            "on_miss must NOT run on a cache HIT"
877        );
878    }
879
880    // ── Test 2: cache MISS runs on_miss, writes back, continues ──
881
882    #[tokio::test]
883    async fn cache_miss_runs_on_miss_sets_continues() {
884        let ttl = Duration::from_secs(30);
885        let repo = Arc::new(MockCacheRepository::new("mock"));
886        let (mut svc, on_miss_invoked) = build_service(
887            repo.clone(),
888            fixed_key(),
889            1024,
890            Some(Body::Bytes(Bytes::from_static(b"x"))),
891            ScriptedOutcome::Complete,
892            Some(ttl),
893            noop_rt(),
894        );
895
896        let outcome = svc.run(exchange()).await;
897
898        let ex = match outcome {
899            PipelineOutcome::Completed(ex) => ex,
900            other => panic!("expected Completed, got {other:?}"),
901        };
902        assert!(on_miss_invoked.load(Ordering::SeqCst));
903        assert_eq!(ex.input.body, Body::Bytes(Bytes::from_static(b"x")));
904        assert_eq!(repo.set_call_count(), 1, "set must be called once on miss");
905        let stored = repo
906            .stored_entry("cache-key")
907            .await
908            .expect("entry must be stored");
909        assert_eq!(stored.bytes, b"x");
910        assert_eq!(stored.content_type, ContentType::Bytes);
911        assert_eq!(repo.last_set_ttl().await, Some(ttl));
912    }
913
914    // ── Test 3: oversized materialized body skips write-back ──
915
916    #[tokio::test]
917    async fn cache_miss_oversized_materialized_body_skips_writeback() {
918        let repo = Arc::new(MockCacheRepository::new("mock"));
919        // max_entry_bytes = 4; on_miss produces 9 bytes.
920        let (mut svc, _invoked) = build_service(
921            repo.clone(),
922            fixed_key(),
923            4,
924            Some(Body::Bytes(Bytes::from_static(b"oversized"))),
925            ScriptedOutcome::Complete,
926            None,
927            noop_rt(),
928        );
929
930        let outcome = svc.run(exchange()).await;
931
932        let ex = match outcome {
933            PipelineOutcome::Completed(ex) => ex,
934            other => panic!("expected Completed, got {other:?}"),
935        };
936        // Body passes through unchanged.
937        assert_eq!(ex.input.body, Body::Bytes(Bytes::from_static(b"oversized")));
938        assert_eq!(
939            repo.set_call_count(),
940            0,
941            "set must NOT be called for oversized body"
942        );
943        assert!(repo.stored_entry("cache-key").await.is_none());
944    }
945
946    // ── Test 4: oversized Stream propagates StreamLimitExceeded ──
947
948    #[tokio::test]
949    async fn cache_miss_oversized_stream_propagates_err() {
950        let repo = Arc::new(MockCacheRepository::new("mock"));
951        let (mut svc, _invoked) = build_service(
952            repo.clone(),
953            fixed_key(),
954            4,
955            Some(stream_body(b"way-too-big-stream")),
956            ScriptedOutcome::Complete,
957            None,
958            noop_rt(),
959        );
960
961        let outcome = svc.run(exchange()).await;
962
963        match outcome {
964            PipelineOutcome::Failed(CamelError::StreamLimitExceeded(n)) => {
965                assert_eq!(n, 4);
966            }
967            other => panic!("expected Failed(StreamLimitExceeded(4)), got {other:?}"),
968        }
969        assert_eq!(
970            repo.set_call_count(),
971            0,
972            "set must NOT be called when stream exceeds limit"
973        );
974    }
975
976    // ── Test 5: on_miss Stopped propagates without write-back ──
977
978    #[tokio::test]
979    async fn cache_on_miss_stopped_propagates_without_writeback() {
980        let repo = Arc::new(MockCacheRepository::new("mock"));
981        let (mut svc, _invoked) = build_service(
982            repo.clone(),
983            fixed_key(),
984            1024,
985            None,
986            ScriptedOutcome::Stop,
987            None,
988            noop_rt(),
989        );
990
991        let outcome = svc.run(exchange()).await;
992
993        assert!(
994            matches!(outcome, PipelineOutcome::Stopped(_)),
995            "Stopped from on_miss MUST propagate as Stopped"
996        );
997        assert_eq!(
998            repo.set_call_count(),
999            0,
1000            "set must NOT be called when on_miss Stops"
1001        );
1002    }
1003
1004    // ── Test 6: on_miss Err propagates without write-back ──
1005
1006    #[tokio::test]
1007    async fn cache_on_miss_err_propagates_without_writeback() {
1008        let repo = Arc::new(MockCacheRepository::new("mock"));
1009        let (mut svc, _invoked) = build_service(
1010            repo.clone(),
1011            fixed_key(),
1012            1024,
1013            None,
1014            ScriptedOutcome::Fail(stub_error("on-miss blew up")),
1015            None,
1016            noop_rt(),
1017        );
1018
1019        let outcome = svc.run(exchange()).await;
1020
1021        match outcome {
1022            PipelineOutcome::Failed(e) => {
1023                assert!(e.to_string().contains("on-miss blew up"), "got: {e}");
1024            }
1025            other => panic!("expected Failed, got {other:?}"),
1026        }
1027        assert_eq!(
1028            repo.set_call_count(),
1029            0,
1030            "set must NOT be called when on_miss fails"
1031        );
1032    }
1033
1034    // ── Test 7: repository get Err propagates ──
1035
1036    #[tokio::test]
1037    async fn cache_repository_get_err_propagates() {
1038        let repo = Arc::new(MockCacheRepository::new("mock"));
1039        repo.set_get_should_fail(true);
1040        let (mut svc, on_miss_invoked) = build_service(
1041            repo,
1042            fixed_key(),
1043            1024,
1044            Some(Body::Bytes(Bytes::from_static(b"x"))),
1045            ScriptedOutcome::Complete,
1046            None,
1047            noop_rt(),
1048        );
1049
1050        let outcome = svc.run(exchange()).await;
1051
1052        match outcome {
1053            PipelineOutcome::Failed(e) => {
1054                assert!(e.to_string().contains("synthetic get failure"), "got: {e}");
1055            }
1056            other => panic!("expected Failed, got {other:?}"),
1057        }
1058        assert!(
1059            !on_miss_invoked.load(Ordering::SeqCst),
1060            "on_miss must NOT run when get fails"
1061        );
1062    }
1063
1064    // ── Test 8: repository set Err propagates ──
1065
1066    #[tokio::test]
1067    async fn cache_repository_set_err_propagates() {
1068        let repo = Arc::new(MockCacheRepository::new("mock"));
1069        repo.set_set_should_fail(true);
1070        let (mut svc, _invoked) = build_service(
1071            repo.clone(),
1072            fixed_key(),
1073            1024,
1074            Some(Body::Bytes(Bytes::from_static(b"x"))),
1075            ScriptedOutcome::Complete,
1076            None,
1077            noop_rt(),
1078        );
1079
1080        let outcome = svc.run(exchange()).await;
1081
1082        match outcome {
1083            PipelineOutcome::Failed(e) => {
1084                assert!(e.to_string().contains("synthetic set failure"), "got: {e}");
1085            }
1086            other => panic!("expected Failed, got {other:?}"),
1087        }
1088        assert_eq!(repo.set_call_count(), 1, "set was attempted (and failed)");
1089    }
1090
1091    // ── Test 9: None key bypasses to on_miss, no set ──
1092
1093    #[tokio::test]
1094    async fn cache_none_key_bypasses_to_on_miss() {
1095        let repo = Arc::new(MockCacheRepository::new("mock"));
1096        let (mut svc, on_miss_invoked) = build_service(
1097            repo.clone(),
1098            none_key(),
1099            1024,
1100            Some(Body::Bytes(Bytes::from_static(b"x"))),
1101            ScriptedOutcome::Complete,
1102            None,
1103            noop_rt(),
1104        );
1105
1106        let outcome = svc.run(exchange()).await;
1107
1108        let ex = match outcome {
1109            PipelineOutcome::Completed(ex) => ex,
1110            other => panic!("expected Completed, got {other:?}"),
1111        };
1112        assert!(
1113            on_miss_invoked.load(Ordering::SeqCst),
1114            "on_miss MUST run when key is None"
1115        );
1116        assert_eq!(ex.input.body, Body::Bytes(Bytes::from_static(b"x")));
1117        assert_eq!(
1118            repo.set_call_count(),
1119            0,
1120            "set must NOT be called when key_expr returns None"
1121        );
1122    }
1123
1124    // ── Extra: HIT reconstruction for each ContentType ──
1125
1126    #[tokio::test]
1127    async fn cache_content_type_reconstruction() {
1128        async fn run_case(entry: CacheEntry, expected: Body) {
1129            let repo = Arc::new(MockCacheRepository::new("mock"));
1130            repo.seed("cache-key", entry).await;
1131            let (mut svc, on_miss_invoked) = build_service(
1132                repo,
1133                fixed_key(),
1134                1024,
1135                Some(Body::Bytes(Bytes::from_static(b"unreached"))),
1136                ScriptedOutcome::Complete,
1137                None,
1138                noop_rt(),
1139            );
1140            let outcome = svc.run(exchange()).await;
1141            let ex = match outcome {
1142                PipelineOutcome::Completed(ex) => ex,
1143                other => panic!("expected Completed, got {other:?}"),
1144            };
1145            assert_eq!(ex.input.body, expected);
1146            assert!(!on_miss_invoked.load(Ordering::SeqCst));
1147        }
1148
1149        run_case(
1150            CacheEntry {
1151                bytes: b"raw".to_vec(),
1152                content_type: ContentType::Bytes,
1153                expires_at: None,
1154            },
1155            Body::Bytes(Bytes::from_static(b"raw")),
1156        )
1157        .await;
1158        run_case(
1159            CacheEntry {
1160                bytes: b"hi".to_vec(),
1161                content_type: ContentType::Text,
1162                expires_at: None,
1163            },
1164            Body::Text("hi".into()),
1165        )
1166        .await;
1167        run_case(
1168            CacheEntry {
1169                bytes: br#"{"k":1}"#.to_vec(),
1170                content_type: ContentType::Json,
1171                expires_at: None,
1172            },
1173            Body::Json(serde_json::json!({"k": 1})),
1174        )
1175        .await;
1176        run_case(
1177            CacheEntry {
1178                bytes: b"<a/>".to_vec(),
1179                content_type: ContentType::Xml,
1180                expires_at: None,
1181            },
1182            Body::Xml("<a/>".into()),
1183        )
1184        .await;
1185    }
1186
1187    // ── Extra: Stream body write-back materializes into Body::Bytes ──
1188
1189    #[tokio::test]
1190    async fn cache_miss_stream_body_is_materialized_and_cached() {
1191        let repo = Arc::new(MockCacheRepository::new("mock"));
1192        let (mut svc, _invoked) = build_service(
1193            repo.clone(),
1194            fixed_key(),
1195            1024,
1196            Some(stream_body(b"chunky")),
1197            ScriptedOutcome::Complete,
1198            None,
1199            noop_rt(),
1200        );
1201
1202        let outcome = svc.run(exchange()).await;
1203
1204        let ex = match outcome {
1205            PipelineOutcome::Completed(ex) => ex,
1206            other => panic!("expected Completed, got {other:?}"),
1207        };
1208        // Stream is replaced by materialized Bytes.
1209        assert_eq!(ex.input.body, Body::Bytes(Bytes::from_static(b"chunky")));
1210        assert_eq!(repo.set_call_count(), 1);
1211        let stored = repo.stored_entry("cache-key").await.expect("stored");
1212        assert_eq!(stored.bytes, b"chunky");
1213        assert_eq!(stored.content_type, ContentType::Bytes);
1214    }
1215
1216    // ── CachePeekStaleService tests ──
1217
1218    #[tokio::test]
1219    async fn cache_peek_stale_serves_post_expiry_entry() {
1220        let repo = Arc::new(MockCacheRepository::new("mock"));
1221        repo.seed(
1222            "cache-key",
1223            CacheEntry {
1224                bytes: b"stale-payload".to_vec(),
1225                content_type: ContentType::Text,
1226                expires_at: None,
1227            },
1228        )
1229        .await;
1230        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1231
1232        let outcome = svc.run(exchange()).await;
1233
1234        let ex = match outcome {
1235            PipelineOutcome::Completed(ex) => ex,
1236            other => panic!("expected Completed, got {other:?}"),
1237        };
1238        assert_eq!(ex.input.body, Body::Text("stale-payload".into()));
1239    }
1240
1241    #[tokio::test]
1242    async fn cache_peek_stale_on_absence_stops_branch() {
1243        let repo = Arc::new(MockCacheRepository::new("mock"));
1244        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1245
1246        let outcome = svc.run(exchange()).await;
1247
1248        assert!(
1249            matches!(outcome, PipelineOutcome::Stopped(_)),
1250            "expected Stopped when no stale entry, got {outcome:?}"
1251        );
1252    }
1253
1254    #[tokio::test]
1255    async fn cache_peek_stale_none_key_stops() {
1256        let repo = Arc::new(MockCacheRepository::new("mock"));
1257        let mut svc = CachePeekStaleService::new(repo, none_key(), PeekStaleMissPolicy::Stop);
1258
1259        let outcome = svc.run(exchange()).await;
1260
1261        assert!(
1262            matches!(outcome, PipelineOutcome::Stopped(_)),
1263            "expected Stopped when key_expr returns None, got {outcome:?}"
1264        );
1265    }
1266
1267    #[tokio::test]
1268    async fn peek_stale_miss_stop_sets_properties_and_stops() {
1269        let repo = Arc::new(MockCacheRepository::new("mock"));
1270        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1271
1272        let outcome = svc.run(exchange()).await;
1273
1274        let ex = match outcome {
1275            PipelineOutcome::Stopped(ex) => ex,
1276            other => panic!("expected Stopped, got {other:?}"),
1277        };
1278        assert_eq!(ex.property(CAMEL_CACHE_PEEK_HIT), Some(&Value::Bool(false)));
1279        assert_eq!(
1280            ex.property(CAMEL_CACHE_PEEK_STALE),
1281            Some(&Value::Bool(false))
1282        );
1283    }
1284
1285    #[tokio::test]
1286    async fn peek_stale_miss_continue_completes_with_body_untouched() {
1287        let repo = Arc::new(MockCacheRepository::new("mock"));
1288        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Continue);
1289
1290        let mut ex = exchange();
1291        ex.input.body = Body::Text("orig".into());
1292
1293        let outcome = svc.run(ex).await;
1294
1295        let ex = match outcome {
1296            PipelineOutcome::Completed(ex) => ex,
1297            other => panic!("expected Completed, got {other:?}"),
1298        };
1299        assert_eq!(ex.input.body, Body::Text("orig".into()));
1300        assert_eq!(ex.property(CAMEL_CACHE_PEEK_HIT), Some(&Value::Bool(false)));
1301        assert_eq!(
1302            ex.property(CAMEL_CACHE_PEEK_STALE),
1303            Some(&Value::Bool(false))
1304        );
1305    }
1306
1307    #[tokio::test]
1308    async fn peek_stale_hit_sets_hit_and_stale_properties() {
1309        let repo = Arc::new(MockCacheRepository::new("mock"));
1310        repo.seed(
1311            "cache-key",
1312            CacheEntry {
1313                bytes: b"stale-payload".to_vec(),
1314                content_type: ContentType::Bytes,
1315                expires_at: Some(SystemTime::now() - Duration::from_millis(1)),
1316            },
1317        )
1318        .await;
1319        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1320
1321        let outcome = svc.run(exchange()).await;
1322
1323        let ex = match outcome {
1324            PipelineOutcome::Completed(ex) => ex,
1325            other => panic!("expected Completed, got {other:?}"),
1326        };
1327        assert_eq!(
1328            ex.input.body,
1329            Body::Bytes(Bytes::from_static(b"stale-payload"))
1330        );
1331        assert_eq!(ex.property(CAMEL_CACHE_PEEK_HIT), Some(&Value::Bool(true)));
1332        assert_eq!(
1333            ex.property(CAMEL_CACHE_PEEK_STALE),
1334            Some(&Value::Bool(true))
1335        );
1336    }
1337
1338    #[tokio::test]
1339    async fn peek_stale_hit_fresh_sets_stale_false() {
1340        let repo = Arc::new(MockCacheRepository::new("mock"));
1341        repo.seed(
1342            "cache-key",
1343            CacheEntry {
1344                bytes: b"fresh-payload".to_vec(),
1345                content_type: ContentType::Bytes,
1346                expires_at: Some(SystemTime::now() + Duration::from_secs(3600)),
1347            },
1348        )
1349        .await;
1350        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1351
1352        let outcome = svc.run(exchange()).await;
1353
1354        let ex = match outcome {
1355            PipelineOutcome::Completed(ex) => ex,
1356            other => panic!("expected Completed, got {other:?}"),
1357        };
1358        assert_eq!(ex.property(CAMEL_CACHE_PEEK_HIT), Some(&Value::Bool(true)));
1359        assert_eq!(
1360            ex.property(CAMEL_CACHE_PEEK_STALE),
1361            Some(&Value::Bool(false))
1362        );
1363    }
1364
1365    // --- Tracing capture helper for debug-log assertions ---
1366
1367    /// `MakeWriter` that appends formatted events to a shared `Vec<u8>` sink.
1368    #[derive(Clone)]
1369    struct CapturingWriter {
1370        sink: Arc<Mutex<Vec<u8>>>,
1371    }
1372
1373    impl std::io::Write for CapturingWriter {
1374        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
1375            self.sink.lock().unwrap().extend_from_slice(buf); // allow-unwrap: test-only
1376            Ok(buf.len())
1377        }
1378        fn flush(&mut self) -> std::io::Result<()> {
1379            Ok(())
1380        }
1381    }
1382
1383    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturingWriter {
1384        type Writer = CapturingWriter;
1385        fn make_writer(&'a self) -> Self::Writer {
1386            self.clone()
1387        }
1388    }
1389
1390    fn debug_sink() -> (Arc<Mutex<Vec<u8>>>, impl tracing::Subscriber) {
1391        let sink: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
1392        let writer = CapturingWriter {
1393            sink: Arc::clone(&sink),
1394        };
1395        let subscriber = tracing_subscriber::fmt()
1396            .with_writer(writer)
1397            .with_ansi(false)
1398            .with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
1399            .finish();
1400        (sink, subscriber)
1401    }
1402
1403    #[tokio::test]
1404    async fn peek_stale_miss_stop_emits_debug_log() {
1405        let repo = Arc::new(MockCacheRepository::new("mock"));
1406        let mut svc = CachePeekStaleService::new(repo, fixed_key(), PeekStaleMissPolicy::Stop);
1407
1408        let (sink, subscriber) = debug_sink();
1409        let _guard = tracing::subscriber::set_default(subscriber);
1410        // Parallel tests race tracing's per-callsite interest cache against
1411        // this thread-local subscriber; force a rebuild so the callsites
1412        // below re-evaluate against it (bd rc-u9hs).
1413        tracing::callsite::rebuild_interest_cache();
1414        let outcome = svc.run(exchange()).await;
1415        drop(_guard);
1416
1417        assert!(matches!(outcome, PipelineOutcome::Stopped(_)));
1418
1419        let captured = String::from_utf8(sink.lock().unwrap().clone()).unwrap(); // allow-unwrap: test-only
1420        let miss_records: Vec<&str> = captured
1421            .lines()
1422            .filter(|l| l.contains("peek miss"))
1423            .collect();
1424        assert_eq!(
1425            miss_records.len(),
1426            1,
1427            "expected exactly one DEBUG record containing \"peek miss\"; got: {captured}"
1428        );
1429        assert!(
1430            miss_records[0].contains("DEBUG"),
1431            "expected DEBUG level record; got: {captured}"
1432        );
1433        assert!(
1434            miss_records[0].contains("repository=mock"),
1435            "expected repository field in record; got: {captured}"
1436        );
1437        assert!(
1438            miss_records[0].contains("step=\"cache_peek_stale\""),
1439            "expected step field in record; got: {captured}"
1440        );
1441    }
1442
1443    #[tokio::test]
1444    async fn peek_stale_key_none_stops_with_debug_log() {
1445        let repo = Arc::new(MockCacheRepository::new("mock"));
1446        let mut svc = CachePeekStaleService::new(repo, none_key(), PeekStaleMissPolicy::Stop);
1447
1448        let (sink, subscriber) = debug_sink();
1449        let _guard = tracing::subscriber::set_default(subscriber);
1450        // Parallel tests race tracing's per-callsite interest cache against
1451        // this thread-local subscriber; force a rebuild so the callsites
1452        // below re-evaluate against it (bd rc-u9hs).
1453        tracing::callsite::rebuild_interest_cache();
1454        let outcome = svc.run(exchange()).await;
1455        drop(_guard);
1456
1457        assert!(matches!(outcome, PipelineOutcome::Stopped(_)));
1458
1459        let captured = String::from_utf8(sink.lock().unwrap().clone()).unwrap(); // allow-unwrap: test-only
1460        let none_records: Vec<&str> = captured
1461            .lines()
1462            .filter(|l| l.contains("resolved to None"))
1463            .collect();
1464        assert_eq!(
1465            none_records.len(),
1466            1,
1467            "expected exactly one DEBUG record containing \"resolved to None\"; got: {captured}"
1468        );
1469        assert!(
1470            none_records[0].contains("DEBUG"),
1471            "expected DEBUG level record; got: {captured}"
1472        );
1473        assert!(
1474            none_records[0].contains("repository=mock"),
1475            "expected repository field in record; got: {captured}"
1476        );
1477        assert!(
1478            none_records[0].contains("step=\"cache_peek_stale\""),
1479            "expected step field in record; got: {captured}"
1480        );
1481    }
1482
1483    // ── CacheInvalidateService tests ──
1484
1485    #[tokio::test]
1486    async fn cache_invalidate_calls_repository_invalidate() {
1487        let repo = Arc::new(MockCacheRepository::new("mock"));
1488        repo.seed(
1489            "cache-key",
1490            CacheEntry {
1491                bytes: b"to-go".to_vec(),
1492                content_type: ContentType::Bytes,
1493                expires_at: None,
1494            },
1495        )
1496        .await;
1497        let mut svc = CacheInvalidateService::new(repo.clone(), fixed_key());
1498
1499        let outcome = svc.run(exchange()).await;
1500
1501        let _ex = match outcome {
1502            PipelineOutcome::Completed(ex) => ex,
1503            other => panic!("expected Completed, got {other:?}"),
1504        };
1505        assert_eq!(
1506            repo.invalidate_call_count(),
1507            1,
1508            "invalidate must be called once"
1509        );
1510        assert_eq!(
1511            repo.last_invalidate_key().await,
1512            Some("cache-key".to_string()),
1513            "invalidate must be called with the correct key"
1514        );
1515        assert!(
1516            repo.stored_entry("cache-key").await.is_none(),
1517            "entry must be removed after invalidation"
1518        );
1519    }
1520
1521    #[tokio::test]
1522    async fn cache_invalidate_none_key_completes() {
1523        let repo = Arc::new(MockCacheRepository::new("mock"));
1524        let mut svc = CacheInvalidateService::new(repo.clone(), none_key());
1525
1526        let outcome = svc.run(exchange()).await;
1527
1528        let _ex = match outcome {
1529            PipelineOutcome::Completed(ex) => ex,
1530            other => panic!("expected Completed, got {other:?}"),
1531        };
1532        assert_eq!(
1533            repo.invalidate_call_count(),
1534            0,
1535            "invalidate must NOT be called when key_expr returns None"
1536        );
1537    }
1538
1539    // ── OTel metrics tests ──
1540
1541    /// Records every `record_counter` call for test assertions.
1542    type CounterRecording = Vec<(String, f64, Vec<(String, String)>)>;
1543
1544    #[derive(Clone)]
1545    struct RecordingMetricsCollector {
1546        counters: Arc<Mutex<CounterRecording>>,
1547    }
1548
1549    impl RecordingMetricsCollector {
1550        fn new() -> Self {
1551            Self {
1552                counters: Arc::new(Mutex::new(Vec::new())),
1553            }
1554        }
1555    }
1556
1557    impl camel_api::metrics::MetricsCollector for RecordingMetricsCollector {
1558        fn record_exchange_duration(&self, _route_id: &str, _duration: Duration) {}
1559        fn increment_errors(&self, _route_id: &str, _error_type: &str) {}
1560        fn increment_exchanges(&self, _route_id: &str) {}
1561        fn set_queue_depth(&self, _route_id: &str, _depth: usize) {}
1562        fn record_circuit_breaker_change(&self, _route_id: &str, _from: &str, _to: &str) {}
1563        fn record_counter(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
1564            self.counters.lock().unwrap().push((
1565                name.to_string(),
1566                value,
1567                labels
1568                    .iter()
1569                    .map(|(k, v)| (k.to_string(), v.to_string()))
1570                    .collect(),
1571            ));
1572        }
1573    }
1574
1575    #[derive(Clone)]
1576    struct TestOtelmRt {
1577        collector: Arc<RecordingMetricsCollector>,
1578    }
1579
1580    impl camel_component_api::health_registry::HealthCheckRegistry for TestOtelmRt {
1581        fn force_unhealthy_for_route(&self, _: &str, _: &str, _: &str) {}
1582    }
1583
1584    impl RuntimeObservability for TestOtelmRt {
1585        fn metrics(&self) -> Arc<dyn camel_api::metrics::MetricsCollector> {
1586            self.collector.clone()
1587        }
1588        fn health(&self) -> Arc<dyn camel_component_api::health_registry::HealthCheckRegistry> {
1589            Arc::new(NoOpHealthCheckRegistry)
1590        }
1591    }
1592
1593    #[tokio::test]
1594    async fn cache_step_hit_increments_otel_counter() {
1595        let repo = Arc::new(MockCacheRepository::new("mock"));
1596        repo.seed(
1597            "cache-key",
1598            CacheEntry {
1599                bytes: b"cached".to_vec(),
1600                content_type: ContentType::Bytes,
1601                expires_at: None,
1602            },
1603        )
1604        .await;
1605        let collector = RecordingMetricsCollector::new();
1606        let counters = collector.counters.clone();
1607        let rt = Arc::new(TestOtelmRt {
1608            collector: Arc::new(collector),
1609        });
1610        let (mut svc, _invoked) = build_service(
1611            repo,
1612            fixed_key(),
1613            1024,
1614            None,
1615            ScriptedOutcome::Complete,
1616            None,
1617            rt,
1618        );
1619
1620        let outcome = svc.run(exchange()).await;
1621        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
1622
1623        let recorded = counters.lock().unwrap().clone();
1624        assert!(
1625            recorded.contains(&(
1626                "camel.cache.hits".to_string(),
1627                1.0,
1628                vec![("repository".to_string(), "mock".to_string())]
1629            )),
1630            "expected camel.cache.hits counter, got: {recorded:?}"
1631        );
1632    }
1633
1634    #[tokio::test]
1635    async fn cache_step_miss_increments_otel_counter() {
1636        let repo = Arc::new(MockCacheRepository::new("mock"));
1637        let collector = RecordingMetricsCollector::new();
1638        let counters = collector.counters.clone();
1639        let rt = Arc::new(TestOtelmRt {
1640            collector: Arc::new(collector),
1641        });
1642        let (mut svc, _invoked) = build_service(
1643            repo.clone(),
1644            fixed_key(),
1645            1024,
1646            Some(Body::Bytes(Bytes::from_static(b"x"))),
1647            ScriptedOutcome::Complete,
1648            None,
1649            rt,
1650        );
1651
1652        let outcome = svc.run(exchange()).await;
1653        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
1654
1655        let recorded = counters.lock().unwrap().clone();
1656        assert!(
1657            recorded.contains(&(
1658                "camel.cache.misses".to_string(),
1659                1.0,
1660                vec![("repository".to_string(), "mock".to_string())]
1661            )),
1662            "expected camel.cache.misses counter, got: {recorded:?}"
1663        );
1664    }
1665}
camel_processor/cache_eip.rs

camel_processor/
cache_eip.rs
