camel_core/step/
function_step.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::sync::Arc;
4use std::task::{Context, Poll};
5use std::time::Duration;
6
7use camel_api::{
8    CamelError, Exchange, ExchangePatch, FunctionDefinition, FunctionId, FunctionInvocationError,
9    FunctionInvoker, PatchBody,
10};
11use tower::Service;
12use tracing::Instrument;
13
14#[derive(Clone)]
15pub struct FunctionStep {
16    definition: FunctionDefinition,
17    invoker: Arc<dyn FunctionInvoker>,
18}
19
20impl FunctionStep {
21    pub fn new(invoker: Arc<dyn FunctionInvoker>, definition: FunctionDefinition) -> Self {
22        Self {
23            definition,
24            invoker,
25        }
26    }
27}
28
29impl Service<Exchange> for FunctionStep {
30    type Response = Exchange;
31    type Error = CamelError;
32    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
33
34    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
35        Poll::Ready(Ok(()))
36    }
37
38    fn call(&mut self, mut ex: Exchange) -> Self::Future {
39        let invoker = Arc::clone(&self.invoker);
40        let id = self.definition.id.clone();
41        let runtime = self.definition.runtime.clone();
42        let timeout_ms = self.definition.timeout_ms;
43        let span = tracing::info_span!(
44            target: "camel_function",
45            "function",
46            function_id = %id.0,
47            runtime = %runtime,
48            timeout_ms = timeout_ms,
49            status = tracing::field::Empty,
50            duration_ms = tracing::field::Empty,
51            error_kind = tracing::field::Empty,
52        );
53        Box::pin(async move {
54            let start = std::time::Instant::now();
55            let outcome: Result<ExchangePatch, CamelError> = async {
56                let result = tokio::time::timeout(
57                    Duration::from_millis(timeout_ms),
58                    invoker.invoke(&id, &ex),
59                )
60                .await
61                .map_err(|_| {
62                    CamelError::ProcessorError(format!(
63                        "function:timeout: {} timed out after {}ms",
64                        id.0, timeout_ms
65                    ))
66                })?;
67                let patch = result.map_err(|e| map_invocation_error(e, &id))?;
68                Ok(patch)
69            }
70            .instrument(span.clone())
71            .await;
72            let elapsed = start.elapsed().as_millis() as u64;
73            span.record("duration_ms", elapsed);
74            match &outcome {
75                Ok(_) => {
76                    span.record("status", "ok");
77                }
78                Err(CamelError::ProcessorError(msg)) => {
79                    let kind = if msg.starts_with("function:timeout:") {
80                        "timeout"
81                    } else if msg.starts_with("function:user_error:") {
82                        "user_error"
83                    } else if msg.starts_with("function:runner_unavailable:") {
84                        "runner_unavailable"
85                    } else if msg.starts_with("function:not_registered:") {
86                        "not_registered"
87                    } else if msg.starts_with("function:transport:") {
88                        "transport"
89                    } else if msg.starts_with("function:invalid_patch:") {
90                        "invalid_patch"
91                    } else {
92                        "unknown"
93                    };
94                    span.record("status", kind);
95                    span.record("error_kind", kind);
96                }
97                Err(_) => {
98                    span.record("status", "unknown");
99                    span.record("error_kind", "unknown");
100                }
101            }
102            let patch = outcome?;
103            apply_patch(&mut ex, patch);
104            Ok(ex)
105        })
106    }
107}
108
109fn map_invocation_error(err: FunctionInvocationError, id: &FunctionId) -> CamelError {
110    match err {
111        FunctionInvocationError::UserError { message, stack, .. } => {
112            let detail = match stack {
113                Some(s) if !s.is_empty() => {
114                    format!("function:user_error: {}: {}\n{}", id.0, message, s)
115                }
116                _ => format!("function:user_error: {}: {}", id.0, message),
117            };
118            CamelError::ProcessorError(detail)
119        }
120        FunctionInvocationError::Timeout { timeout_ms, .. } => CamelError::ProcessorError(format!(
121            "function:timeout: {} timed out after {}ms",
122            id.0, timeout_ms
123        )),
124        FunctionInvocationError::NotRegistered { .. } => {
125            CamelError::ProcessorError(format!("function:not_registered: {}", id.0))
126        }
127        FunctionInvocationError::RunnerUnavailable { reason } => {
128            CamelError::ProcessorError(format!("function:runner_unavailable: {}: {}", id.0, reason))
129        }
130        FunctionInvocationError::Transport(msg) => {
131            CamelError::ProcessorError(format!("function:transport: {}: {}", id.0, msg))
132        }
133        FunctionInvocationError::InvalidPatch(msg) => {
134            CamelError::ProcessorError(format!("function:invalid_patch: {}: {}", id.0, msg))
135        }
136        _ => CamelError::ProcessorError(format!(
137            "function:error: {}: unknown invocation error",
138            id.0
139        )),
140    }
141}
142
143fn apply_patch(ex: &mut Exchange, patch: ExchangePatch) {
144    if let Some(body) = patch.body {
145        ex.input.body = match body {
146            PatchBody::Text(s) => s.into(),
147            PatchBody::Json(v) => v.into(),
148            PatchBody::Empty => camel_api::Body::Empty,
149            // Future PatchBody variants leave the body unchanged.
150            _ => ex.input.body.clone(),
151        };
152    }
153    for (k, v) in patch.headers_set {
154        ex.input.headers.insert(k, v);
155    }
156    for k in patch.headers_removed {
157        ex.input.headers.remove(&k);
158    }
159    for (k, v) in patch.properties_set {
160        ex.properties.insert(k, v);
161    }
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use async_trait::async_trait;
168    use camel_api::function::PrepareToken;
169    use camel_api::{FunctionDiff, FunctionInvokerSync};
170    use std::sync::Mutex;
171
172    struct MockInvoker {
173        responses: Mutex<Vec<Result<ExchangePatch, FunctionInvocationError>>>,
174    }
175
176    impl MockInvoker {
177        fn new(responses: Vec<Result<ExchangePatch, FunctionInvocationError>>) -> Self {
178            Self {
179                responses: Mutex::new(responses),
180            }
181        }
182    }
183
184    impl FunctionInvokerSync for MockInvoker {
185        fn stage_pending(
186            &self,
187            _def: FunctionDefinition,
188            _route_id: Option<&str>,
189            _generation: u64,
190        ) {
191        }
192        fn discard_staging(&self, _generation: u64) {}
193        fn begin_reload(&self) -> u64 {
194            0
195        }
196        fn function_refs_for_route(&self, _route_id: &str) -> Vec<(FunctionId, Option<String>)> {
197            vec![]
198        }
199        fn staged_refs_for_route(
200            &self,
201            _route_id: &str,
202            _generation: u64,
203        ) -> Vec<(FunctionId, Option<String>)> {
204            vec![]
205        }
206        fn staged_defs_for_route(
207            &self,
208            _route_id: &str,
209            _generation: u64,
210        ) -> Vec<(FunctionDefinition, Option<String>)> {
211            vec![]
212        }
213    }
214
215    #[async_trait]
216    impl FunctionInvoker for MockInvoker {
217        async fn register(
218            &self,
219            _def: FunctionDefinition,
220            _route_id: Option<&str>,
221        ) -> Result<(), FunctionInvocationError> {
222            Ok(())
223        }
224        async fn unregister(
225            &self,
226            _id: &FunctionId,
227            _route_id: Option<&str>,
228        ) -> Result<(), FunctionInvocationError> {
229            Ok(())
230        }
231        async fn invoke(
232            &self,
233            _id: &FunctionId,
234            _exchange: &Exchange,
235        ) -> Result<ExchangePatch, FunctionInvocationError> {
236            let mut resp = self.responses.lock().unwrap();
237            resp.remove(0)
238        }
239        async fn prepare_reload(
240            &self,
241            _diff: FunctionDiff,
242            _generation: u64,
243        ) -> Result<PrepareToken, FunctionInvocationError> {
244            Ok(PrepareToken::default())
245        }
246        async fn finalize_reload(
247            &self,
248            _diff: &FunctionDiff,
249            _generation: u64,
250        ) -> Result<(), FunctionInvocationError> {
251            Ok(())
252        }
253        async fn rollback_reload(
254            &self,
255            _token: PrepareToken,
256            _generation: u64,
257        ) -> Result<(), FunctionInvocationError> {
258            Ok(())
259        }
260        async fn commit_reload(
261            &self,
262            _diff: FunctionDiff,
263            _generation: u64,
264        ) -> Result<(), FunctionInvocationError> {
265            Ok(())
266        }
267        async fn commit_staged(&self) -> Result<(), FunctionInvocationError> {
268            Ok(())
269        }
270    }
271
272    fn test_definition() -> FunctionDefinition {
273        FunctionDefinition {
274            id: FunctionId::compute("deno", "test", 5000),
275            runtime: "deno".into(),
276            source: "test".into(),
277            timeout_ms: 5000,
278            route_id: None,
279            step_index: None,
280        }
281    }
282
283    #[tokio::test]
284    async fn function_step_applies_patch_body_text() {
285        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
286            body: Some(PatchBody::Text("patched".into())),
287            ..Default::default()
288        })]));
289        let mut step = FunctionStep::new(invoker, test_definition());
290        let ex = Exchange::default();
291        let result = step.call(ex).await.unwrap();
292        assert_eq!(result.input.body.as_text(), Some("patched"));
293    }
294
295    #[tokio::test]
296    async fn function_step_applies_patch_headers() {
297        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
298            headers_set: vec![("x-key".into(), serde_json::json!("val"))],
299            headers_removed: vec!["x-old".into()],
300            ..Default::default()
301        })]));
302        let mut step = FunctionStep::new(invoker, test_definition());
303        let mut ex = Exchange::default();
304        ex.input
305            .headers
306            .insert("x-old".into(), serde_json::json!("gone"));
307        let result = step.call(ex).await.unwrap();
308        assert_eq!(
309            result.input.headers.get("x-key").unwrap().as_str(),
310            Some("val")
311        );
312        assert!(!result.input.headers.contains_key("x-old"));
313    }
314
315    #[tokio::test]
316    async fn function_step_applies_patch_properties() {
317        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
318            properties_set: vec![("prop".into(), serde_json::json!(42))],
319            ..Default::default()
320        })]));
321        let mut step = FunctionStep::new(invoker, test_definition());
322        let ex = Exchange::default();
323        let result = step.call(ex).await.unwrap();
324        assert_eq!(result.properties.get("prop").unwrap().as_i64(), Some(42));
325    }
326
327    #[tokio::test]
328    async fn function_step_maps_timeout_error() {
329        let invoker = Arc::new(MockInvoker::new(vec![Err(
330            FunctionInvocationError::Timeout {
331                function_id: FunctionId("x".into()),
332                timeout_ms: 5000,
333            },
334        )]));
335        let mut step = FunctionStep::new(invoker, test_definition());
336        let ex = Exchange::default();
337        let err = step.call(ex).await.unwrap_err();
338        let msg = match &err {
339            CamelError::ProcessorError(m) => m,
340            _ => panic!("wrong error type"),
341        };
342        assert!(msg.contains("function:timeout:"));
343    }
344
345    #[tokio::test]
346    async fn function_step_maps_user_error() {
347        let invoker = Arc::new(MockInvoker::new(vec![Err(
348            FunctionInvocationError::UserError {
349                function_id: FunctionId("x".into()),
350                message: "boom".into(),
351                stack: None,
352            },
353        )]));
354        let mut step = FunctionStep::new(invoker, test_definition());
355        let ex = Exchange::default();
356        let err = step.call(ex).await.unwrap_err();
357        let msg = match &err {
358            CamelError::ProcessorError(m) => m,
359            _ => panic!("wrong error type"),
360        };
361        assert!(msg.contains("function:user_error:"));
362        assert!(msg.contains("boom"));
363    }
364
365    #[tokio::test]
366    async fn function_step_client_side_timeout_fires() {
367        struct SlowInvoker;
368        impl FunctionInvokerSync for SlowInvoker {
369            fn stage_pending(
370                &self,
371                _def: FunctionDefinition,
372                _route_id: Option<&str>,
373                _generation: u64,
374            ) {
375            }
376            fn discard_staging(&self, _generation: u64) {}
377            fn begin_reload(&self) -> u64 {
378                0
379            }
380            fn function_refs_for_route(
381                &self,
382                _route_id: &str,
383            ) -> Vec<(FunctionId, Option<String>)> {
384                vec![]
385            }
386            fn staged_refs_for_route(
387                &self,
388                _route_id: &str,
389                _generation: u64,
390            ) -> Vec<(FunctionId, Option<String>)> {
391                vec![]
392            }
393            fn staged_defs_for_route(
394                &self,
395                _route_id: &str,
396                _generation: u64,
397            ) -> Vec<(FunctionDefinition, Option<String>)> {
398                vec![]
399            }
400        }
401        #[async_trait]
402        impl FunctionInvoker for SlowInvoker {
403            async fn register(
404                &self,
405                _def: FunctionDefinition,
406                _route_id: Option<&str>,
407            ) -> Result<(), FunctionInvocationError> {
408                Ok(())
409            }
410            async fn unregister(
411                &self,
412                _id: &FunctionId,
413                _route_id: Option<&str>,
414            ) -> Result<(), FunctionInvocationError> {
415                Ok(())
416            }
417            async fn invoke(
418                &self,
419                _id: &FunctionId,
420                _exchange: &Exchange,
421            ) -> Result<ExchangePatch, FunctionInvocationError> {
422                tokio::time::sleep(Duration::from_secs(10)).await;
423                Ok(ExchangePatch::default())
424            }
425            async fn prepare_reload(
426                &self,
427                _diff: FunctionDiff,
428                _generation: u64,
429            ) -> Result<PrepareToken, FunctionInvocationError> {
430                Ok(PrepareToken::default())
431            }
432            async fn finalize_reload(
433                &self,
434                _diff: &FunctionDiff,
435                _generation: u64,
436            ) -> Result<(), FunctionInvocationError> {
437                Ok(())
438            }
439            async fn rollback_reload(
440                &self,
441                _token: PrepareToken,
442                _generation: u64,
443            ) -> Result<(), FunctionInvocationError> {
444                Ok(())
445            }
446            async fn commit_reload(
447                &self,
448                _diff: FunctionDiff,
449                _generation: u64,
450            ) -> Result<(), FunctionInvocationError> {
451                Ok(())
452            }
453            async fn commit_staged(&self) -> Result<(), FunctionInvocationError> {
454                Ok(())
455            }
456        }
457        let def = FunctionDefinition {
458            id: FunctionId::compute("deno", "slow", 50),
459            runtime: "deno".into(),
460            source: "slow".into(),
461            timeout_ms: 50,
462            route_id: None,
463            step_index: None,
464        };
465        let mut step = FunctionStep::new(Arc::new(SlowInvoker), def);
466        let ex = Exchange::default();
467        let err = step.call(ex).await.unwrap_err();
468        let msg = match &err {
469            CamelError::ProcessorError(m) => m,
470            _ => panic!("wrong error type"),
471        };
472        assert!(msg.contains("function:timeout:"));
473    }
474
475    #[tokio::test]
476    async fn function_step_emits_tracing_span() {
477        use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
478        use tracing_subscriber::prelude::*;
479
480        // OnceLock-gated global registry: heals/prevents callsite-interest
481        // poisoning of the shared `function` span callsite (target
482        // `camel_function`), which subscriber-less sibling function-step
483        // tests in this binary hit first (fix pattern: c3853198; bd
484        // rc-img5).
485        static INIT: std::sync::OnceLock<()> = std::sync::OnceLock::new();
486        if INIT.set(()).is_ok() {
487            let _ = tracing::subscriber::set_global_default(tracing_subscriber::registry());
488        }
489
490        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch::default())]));
491        let def = FunctionDefinition {
492            id: FunctionId::compute("deno", "span_test", 5000),
493            runtime: "deno".into(),
494            source: "span_test".into(),
495            timeout_ms: 5000,
496            route_id: None,
497            step_index: None,
498        };
499        let mut step = FunctionStep::new(invoker, def);
500        let ex = Exchange::default();
501
502        let span_seen = Arc::new(AtomicBool::new(false));
503        let span_seen_clone = span_seen.clone();
504
505        let layer = tracing_subscriber::fmt::layer()
506            .with_writer(std::io::sink)
507            .with_filter(tracing_subscriber::filter::filter_fn(move |meta| {
508                if meta.target() == "camel_function" && meta.name() == "function" {
509                    span_seen_clone.store(true, AtomicOrdering::SeqCst);
510                }
511                true
512            }));
513
514        let _guard = tracing_subscriber::registry().with(layer).set_default();
515        let result = step.call(ex).await;
516        assert!(result.is_ok());
517        assert!(
518            span_seen.load(AtomicOrdering::SeqCst),
519            "expected function span with target 'camel_function' and name 'function'"
520        );
521    }
522
523    #[tokio::test]
524    async fn function_step_user_error_with_stack() {
525        let invoker = Arc::new(MockInvoker::new(vec![Err(
526            FunctionInvocationError::UserError {
527                function_id: FunctionId("x".into()),
528                message: "custom error".into(),
529                stack: Some("at line 1\nat line 2".into()),
530            },
531        )]));
532        let mut step = FunctionStep::new(invoker, test_definition());
533        let ex = Exchange::default();
534        let err = step.call(ex).await.unwrap_err();
535        let msg = match &err {
536            CamelError::ProcessorError(m) => m.clone(),
537            _ => panic!("wrong error type"),
538        };
539        assert!(msg.contains("function:user_error:"));
540        assert!(msg.contains("custom error"));
541        assert!(msg.contains("at line 1"));
542    }
543
544    #[tokio::test]
545    async fn function_step_maps_not_registered_error() {
546        let invoker = Arc::new(MockInvoker::new(vec![Err(
547            FunctionInvocationError::NotRegistered {
548                function_id: FunctionId("missing-fn".into()),
549            },
550        )]));
551        let mut step = FunctionStep::new(invoker, test_definition());
552        let ex = Exchange::default();
553        let err = step.call(ex).await.unwrap_err();
554        let msg = match &err {
555            CamelError::ProcessorError(m) => m,
556            other => panic!("wrong error type: {:?}", other),
557        };
558        assert!(msg.contains("function:not_registered:"), "msg was: {}", msg);
559    }
560
561    #[tokio::test]
562    async fn function_step_maps_runner_unavailable_error() {
563        let invoker = Arc::new(MockInvoker::new(vec![Err(
564            FunctionInvocationError::RunnerUnavailable {
565                reason: "runtime crashed".into(),
566            },
567        )]));
568        let mut step = FunctionStep::new(invoker, test_definition());
569        let ex = Exchange::default();
570        let err = step.call(ex).await.unwrap_err();
571        let msg = match &err {
572            CamelError::ProcessorError(m) => m,
573            _ => panic!("wrong error type"),
574        };
575        assert!(msg.contains("function:runner_unavailable:"));
576        assert!(msg.contains("runtime crashed"));
577    }
578
579    #[tokio::test]
580    async fn function_step_maps_transport_error() {
581        let invoker = Arc::new(MockInvoker::new(vec![Err(
582            FunctionInvocationError::Transport("connection refused".into()),
583        )]));
584        let mut step = FunctionStep::new(invoker, test_definition());
585        let ex = Exchange::default();
586        let err = step.call(ex).await.unwrap_err();
587        let msg = match &err {
588            CamelError::ProcessorError(m) => m,
589            _ => panic!("wrong error type"),
590        };
591        assert!(msg.contains("function:transport:"));
592        assert!(msg.contains("connection refused"));
593    }
594
595    #[tokio::test]
596    async fn function_step_maps_invalid_patch_error() {
597        let invoker = Arc::new(MockInvoker::new(vec![Err(
598            FunctionInvocationError::InvalidPatch("missing field 'body'".into()),
599        )]));
600        let mut step = FunctionStep::new(invoker, test_definition());
601        let ex = Exchange::default();
602        let err = step.call(ex).await.unwrap_err();
603        let msg = match &err {
604            CamelError::ProcessorError(m) => m,
605            _ => panic!("wrong error type"),
606        };
607        assert!(msg.contains("function:invalid_patch:"));
608        assert!(msg.contains("missing field 'body'"));
609    }
610
611    #[tokio::test]
612    async fn function_step_applies_patch_body_json() {
613        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
614            body: Some(PatchBody::Json(serde_json::json!({"key": "value"}))),
615            ..Default::default()
616        })]));
617        let mut step = FunctionStep::new(invoker, test_definition());
618        let ex = Exchange::default();
619        let result = step.call(ex).await.unwrap();
620        match result.input.body {
621            camel_api::Body::Json(v) => assert_eq!(v.get("key").unwrap().as_str(), Some("value")),
622            other => panic!("expected Json body, got {:?}", other),
623        }
624    }
625
626    #[tokio::test]
627    async fn function_step_applies_patch_body_empty() {
628        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
629            body: Some(PatchBody::Empty),
630            ..Default::default()
631        })]));
632        let mut step = FunctionStep::new(invoker, test_definition());
633        let ex = Exchange::default();
634        let result = step.call(ex).await.unwrap();
635        assert!(matches!(result.input.body, camel_api::Body::Empty));
636    }
637
638    #[tokio::test]
639    async fn function_step_poll_ready_always_ready() {
640        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch::default())]));
641        let mut step = FunctionStep::new(invoker, test_definition());
642        let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
643        let poll = step.poll_ready(&mut cx);
644        assert!(matches!(poll, std::task::Poll::Ready(Ok(()))));
645    }
646
647    #[tokio::test]
648    async fn function_step_user_error_with_empty_stack() {
649        let invoker = Arc::new(MockInvoker::new(vec![Err(
650            FunctionInvocationError::UserError {
651                function_id: FunctionId("x".into()),
652                message: "no stack".into(),
653                stack: Some("".into()),
654            },
655        )]));
656        let mut step = FunctionStep::new(invoker, test_definition());
657        let ex = Exchange::default();
658        let err = step.call(ex).await.unwrap_err();
659        let msg = match &err {
660            CamelError::ProcessorError(m) => m,
661            _ => panic!("wrong error type"),
662        };
663        assert!(msg.contains("function:user_error:"));
664        assert!(msg.contains("no stack"));
665        assert!(!msg.contains("\n"));
666    }
667
668    #[tokio::test]
669    async fn function_step_preserves_exchange_properties_not_in_patch() {
670        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
671            body: Some(PatchBody::Text("new body".into())),
672            ..Default::default()
673        })]));
674        let mut step = FunctionStep::new(invoker, test_definition());
675        let mut ex = Exchange::default();
676        ex.properties
677            .insert("existing".into(), serde_json::json!("keep"));
678        let result = step.call(ex).await.unwrap();
679        assert_eq!(
680            result.properties.get("existing").unwrap().as_str(),
681            Some("keep")
682        );
683        assert_eq!(result.input.body.as_text(), Some("new body"));
684    }
685
686    #[tokio::test]
687    async fn function_step_removes_header_not_in_set() {
688        let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
689            headers_removed: vec!["to-remove".into()],
690            ..Default::default()
691        })]));
692        let mut step = FunctionStep::new(invoker, test_definition());
693        let mut ex = Exchange::default();
694        ex.input
695            .headers
696            .insert("to-remove".into(), serde_json::json!("old"));
697        ex.input
698            .headers
699            .insert("to-keep".into(), serde_json::json!("stay"));
700        let result = step.call(ex).await.unwrap();
701        assert!(!result.input.headers.contains_key("to-remove"));
702        assert!(result.input.headers.contains_key("to-keep"));
703    }
704}
camel_core/step/function_step.rs

camel_core/step/
function_step.rs
