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 _ => 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 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch::default())]));
481 let def = FunctionDefinition {
482 id: FunctionId::compute("deno", "span_test", 5000),
483 runtime: "deno".into(),
484 source: "span_test".into(),
485 timeout_ms: 5000,
486 route_id: None,
487 step_index: None,
488 };
489 let mut step = FunctionStep::new(invoker, def);
490 let ex = Exchange::default();
491
492 let span_seen = Arc::new(AtomicBool::new(false));
493 let span_seen_clone = span_seen.clone();
494
495 let layer = tracing_subscriber::fmt::layer()
496 .with_writer(std::io::sink)
497 .with_filter(tracing_subscriber::filter::filter_fn(move |meta| {
498 if meta.target() == "camel_function" && meta.name() == "function" {
499 span_seen_clone.store(true, AtomicOrdering::SeqCst);
500 }
501 true
502 }));
503
504 let _guard = tracing_subscriber::registry().with(layer).set_default();
505 let result = step.call(ex).await;
506 assert!(result.is_ok());
507 assert!(
508 span_seen.load(AtomicOrdering::SeqCst),
509 "expected function span with target 'camel_function' and name 'function'"
510 );
511 }
512
513 #[tokio::test]
514 async fn function_step_user_error_with_stack() {
515 let invoker = Arc::new(MockInvoker::new(vec![Err(
516 FunctionInvocationError::UserError {
517 function_id: FunctionId("x".into()),
518 message: "custom error".into(),
519 stack: Some("at line 1\nat line 2".into()),
520 },
521 )]));
522 let mut step = FunctionStep::new(invoker, test_definition());
523 let ex = Exchange::default();
524 let err = step.call(ex).await.unwrap_err();
525 let msg = match &err {
526 CamelError::ProcessorError(m) => m.clone(),
527 _ => panic!("wrong error type"),
528 };
529 assert!(msg.contains("function:user_error:"));
530 assert!(msg.contains("custom error"));
531 assert!(msg.contains("at line 1"));
532 }
533
534 #[tokio::test]
535 async fn function_step_maps_not_registered_error() {
536 let invoker = Arc::new(MockInvoker::new(vec![Err(
537 FunctionInvocationError::NotRegistered {
538 function_id: FunctionId("missing-fn".into()),
539 },
540 )]));
541 let mut step = FunctionStep::new(invoker, test_definition());
542 let ex = Exchange::default();
543 let err = step.call(ex).await.unwrap_err();
544 let msg = match &err {
545 CamelError::ProcessorError(m) => m,
546 other => panic!("wrong error type: {:?}", other),
547 };
548 assert!(msg.contains("function:not_registered:"), "msg was: {}", msg);
549 }
550
551 #[tokio::test]
552 async fn function_step_maps_runner_unavailable_error() {
553 let invoker = Arc::new(MockInvoker::new(vec![Err(
554 FunctionInvocationError::RunnerUnavailable {
555 reason: "runtime crashed".into(),
556 },
557 )]));
558 let mut step = FunctionStep::new(invoker, test_definition());
559 let ex = Exchange::default();
560 let err = step.call(ex).await.unwrap_err();
561 let msg = match &err {
562 CamelError::ProcessorError(m) => m,
563 _ => panic!("wrong error type"),
564 };
565 assert!(msg.contains("function:runner_unavailable:"));
566 assert!(msg.contains("runtime crashed"));
567 }
568
569 #[tokio::test]
570 async fn function_step_maps_transport_error() {
571 let invoker = Arc::new(MockInvoker::new(vec![Err(
572 FunctionInvocationError::Transport("connection refused".into()),
573 )]));
574 let mut step = FunctionStep::new(invoker, test_definition());
575 let ex = Exchange::default();
576 let err = step.call(ex).await.unwrap_err();
577 let msg = match &err {
578 CamelError::ProcessorError(m) => m,
579 _ => panic!("wrong error type"),
580 };
581 assert!(msg.contains("function:transport:"));
582 assert!(msg.contains("connection refused"));
583 }
584
585 #[tokio::test]
586 async fn function_step_maps_invalid_patch_error() {
587 let invoker = Arc::new(MockInvoker::new(vec![Err(
588 FunctionInvocationError::InvalidPatch("missing field 'body'".into()),
589 )]));
590 let mut step = FunctionStep::new(invoker, test_definition());
591 let ex = Exchange::default();
592 let err = step.call(ex).await.unwrap_err();
593 let msg = match &err {
594 CamelError::ProcessorError(m) => m,
595 _ => panic!("wrong error type"),
596 };
597 assert!(msg.contains("function:invalid_patch:"));
598 assert!(msg.contains("missing field 'body'"));
599 }
600
601 #[tokio::test]
602 async fn function_step_applies_patch_body_json() {
603 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
604 body: Some(PatchBody::Json(serde_json::json!({"key": "value"}))),
605 ..Default::default()
606 })]));
607 let mut step = FunctionStep::new(invoker, test_definition());
608 let ex = Exchange::default();
609 let result = step.call(ex).await.unwrap();
610 match result.input.body {
611 camel_api::Body::Json(v) => assert_eq!(v.get("key").unwrap().as_str(), Some("value")),
612 other => panic!("expected Json body, got {:?}", other),
613 }
614 }
615
616 #[tokio::test]
617 async fn function_step_applies_patch_body_empty() {
618 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
619 body: Some(PatchBody::Empty),
620 ..Default::default()
621 })]));
622 let mut step = FunctionStep::new(invoker, test_definition());
623 let ex = Exchange::default();
624 let result = step.call(ex).await.unwrap();
625 assert!(matches!(result.input.body, camel_api::Body::Empty));
626 }
627
628 #[tokio::test]
629 async fn function_step_poll_ready_always_ready() {
630 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch::default())]));
631 let mut step = FunctionStep::new(invoker, test_definition());
632 let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
633 let poll = step.poll_ready(&mut cx);
634 assert!(matches!(poll, std::task::Poll::Ready(Ok(()))));
635 }
636
637 #[tokio::test]
638 async fn function_step_user_error_with_empty_stack() {
639 let invoker = Arc::new(MockInvoker::new(vec![Err(
640 FunctionInvocationError::UserError {
641 function_id: FunctionId("x".into()),
642 message: "no stack".into(),
643 stack: Some("".into()),
644 },
645 )]));
646 let mut step = FunctionStep::new(invoker, test_definition());
647 let ex = Exchange::default();
648 let err = step.call(ex).await.unwrap_err();
649 let msg = match &err {
650 CamelError::ProcessorError(m) => m,
651 _ => panic!("wrong error type"),
652 };
653 assert!(msg.contains("function:user_error:"));
654 assert!(msg.contains("no stack"));
655 assert!(!msg.contains("\n"));
656 }
657
658 #[tokio::test]
659 async fn function_step_preserves_exchange_properties_not_in_patch() {
660 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
661 body: Some(PatchBody::Text("new body".into())),
662 ..Default::default()
663 })]));
664 let mut step = FunctionStep::new(invoker, test_definition());
665 let mut ex = Exchange::default();
666 ex.properties
667 .insert("existing".into(), serde_json::json!("keep"));
668 let result = step.call(ex).await.unwrap();
669 assert_eq!(
670 result.properties.get("existing").unwrap().as_str(),
671 Some("keep")
672 );
673 assert_eq!(result.input.body.as_text(), Some("new body"));
674 }
675
676 #[tokio::test]
677 async fn function_step_removes_header_not_in_set() {
678 let invoker = Arc::new(MockInvoker::new(vec![Ok(ExchangePatch {
679 headers_removed: vec!["to-remove".into()],
680 ..Default::default()
681 })]));
682 let mut step = FunctionStep::new(invoker, test_definition());
683 let mut ex = Exchange::default();
684 ex.input
685 .headers
686 .insert("to-remove".into(), serde_json::json!("old"));
687 ex.input
688 .headers
689 .insert("to-keep".into(), serde_json::json!("stay"));
690 let result = step.call(ex).await.unwrap();
691 assert!(!result.input.headers.contains_key("to-remove"));
692 assert!(result.input.headers.contains_key("to-keep"));
693 }
694}