1use crate::retries::RetryPolicy;
2use async_trait::async_trait;
3use chrono::{DateTime, Utc};
4use platform_core::{
5 ActorContext, AppError, AppResult, CorrelationId, DbPool, ErrorCode, ExecutionContext,
6 ExecutionId, RuntimeSpanAttributes, TenantId, TraceContext, db::DbTransaction,
7 record_runtime_span_attributes, trace_context_from_headers, trace_headers,
8};
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11use std::collections::BTreeMap;
12use std::fmt::Debug;
13use std::sync::Arc;
14use std::time::{Duration, Instant};
15use tracing::Instrument;
16use uuid::Uuid;
17
18const STALE_PROCESSING_LOCK_SECONDS: i64 = 300;
19
20#[async_trait]
21pub trait FunctionHandler: Debug + Send + Sync {
22 async fn call(&self, ctx: ExecutionContext, input: Value) -> AppResult<Value>;
23
24 fn observability(&self) -> Option<FunctionHandlerObservability> {
25 None
26 }
27}
28
29pub use FunctionHandler as RuntimeFunction;
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct FunctionHandlerObservability {
33 pub source: String,
34 pub attributes: Value,
35}
36
37impl FunctionHandlerObservability {
38 pub fn new(source: impl Into<String>, attributes: Value) -> Self {
39 Self {
40 source: source.into(),
41 attributes: normalize_log_attributes(attributes),
42 }
43 }
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47enum ExecutionLogSeverity {
48 Info,
49 Error,
50}
51
52impl ExecutionLogSeverity {
53 fn as_str(self) -> &'static str {
54 match self {
55 Self::Info => "info",
56 Self::Error => "error",
57 }
58 }
59}
60
61#[derive(Debug, Clone)]
62pub struct FunctionDefinition {
63 pub name: String,
64 pub version: u16,
65 pub queue: String,
66 pub retry_policy: RetryPolicy,
67 pub handler: Arc<dyn FunctionHandler>,
68}
69
70#[derive(Debug, Default, Clone)]
71pub struct FunctionRegistry {
72 functions: BTreeMap<String, FunctionDefinition>,
73}
74
75impl FunctionRegistry {
76 pub fn register(&mut self, function: FunctionDefinition) {
77 self.functions.insert(function.name.clone(), function);
78 }
79
80 pub fn get(&self, name: &str) -> Option<&FunctionDefinition> {
81 self.functions.get(name)
82 }
83
84 pub fn all(&self) -> impl Iterator<Item = &FunctionDefinition> {
85 self.functions.values()
86 }
87}
88
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
90#[serde(rename_all = "snake_case")]
91pub enum FunctionRunStatus {
92 Pending,
93 Processing,
94 Completed,
95 Failed,
96 Dead,
97}
98
99#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
100#[serde(rename_all = "snake_case")]
101pub enum FunctionTenancyMode {
102 None,
103 Optional,
104 Required,
105}
106
107impl FunctionTenancyMode {
108 const fn as_str(self) -> &'static str {
109 match self {
110 Self::None => "none",
111 Self::Optional => "optional",
112 Self::Required => "required",
113 }
114 }
115
116 const fn accepts(self, tenant_id: Option<&TenantId>) -> bool {
117 match self {
118 Self::None => tenant_id.is_none(),
119 Self::Optional => true,
120 Self::Required => tenant_id.is_some(),
121 }
122 }
123}
124
125impl FunctionRunStatus {
126 fn as_str(self) -> &'static str {
127 match self {
128 Self::Pending => "pending",
129 Self::Processing => "processing",
130 Self::Completed => "completed",
131 Self::Failed => "failed",
132 Self::Dead => "dead",
133 }
134 }
135}
136
137#[derive(Debug, Clone)]
138pub struct EnqueueFunctionRequest {
139 pub function_name: String,
140 pub input_json: Value,
141 pub correlation_id: CorrelationId,
142 pub actor: ActorContext,
143 pub tenant_id: Option<TenantId>,
144 pub tenancy_mode: FunctionTenancyMode,
145 pub trace: TraceContext,
146 pub causation_id: Option<String>,
147 pub max_attempts: Option<i32>,
148}
149
150#[derive(Debug, Clone, Deserialize, Serialize)]
151pub struct ClaimedFunctionRun {
152 pub id: String,
153 pub function_name: String,
154 pub input_json: Value,
155 pub attempts: i32,
156 pub max_attempts: i32,
157 pub correlation_id: String,
158 pub actor: ActorContext,
159 pub tenant_id: Option<TenantId>,
160 pub tenancy_mode: FunctionTenancyMode,
161 pub trace: TraceContext,
162 pub causation_id: Option<String>,
163}
164
165#[derive(Debug, Clone)]
166pub struct RuntimeClient {
167 pool: DbPool,
168}
169
170impl RuntimeClient {
171 pub fn new(pool: DbPool) -> Self {
172 Self { pool }
173 }
174
175 pub async fn enqueue_function(&self, request: EnqueueFunctionRequest) -> AppResult<String> {
176 let mut tx = self.pool.begin().await.map_err(map_runtime_error)?;
177 let run = self.enqueue_function_in_tx(&mut tx, request).await?;
178 tx.commit().await.map_err(map_runtime_error)?;
179 self.record_function_enqueued(&run).await;
180 Ok(run.id)
181 }
182
183 pub(crate) async fn enqueue_function_in_tx(
184 &self,
185 tx: &mut DbTransaction<'_>,
186 request: EnqueueFunctionRequest,
187 ) -> AppResult<EnqueuedFunctionRun> {
188 if !request.tenancy_mode.accepts(request.tenant_id.as_ref()) {
189 return Err(AppError::new(
190 ErrorCode::Validation,
191 "function tenant context is incompatible with its tenancy mode",
192 ));
193 }
194 let id = format!("fnrun_{}", Uuid::now_v7());
195 let max_attempts = request.max_attempts.unwrap_or(3);
196 let mut input_json = request.input_json;
197 attach_runtime_context_to_input(
198 &mut input_json,
199 &request.correlation_id,
200 &request.trace,
201 request.causation_id.as_deref(),
202 );
203 let span = tracing::info_span!(
204 "function_enqueue",
205 lenso.correlation_id = tracing::field::Empty,
206 lenso.story_id = tracing::field::Empty,
207 lenso.function_run_id = tracing::field::Empty,
208 lenso.execution.kind = tracing::field::Empty,
209 lenso.execution.name = tracing::field::Empty,
210 );
211 record_runtime_span_attributes(
212 &span,
213 &RuntimeSpanAttributes::function(
214 request.correlation_id.0.clone(),
215 id.clone(),
216 request.function_name.clone(),
217 ),
218 );
219
220 async {
221 sqlx::query(
222 r#"
223 insert into runtime.function_runs (
224 id,
225 function_name,
226 input_json,
227 max_attempts,
228 correlation_id,
229 actor,
230 tenant_id,
231 tenancy_mode
232 )
233 values ($1, $2, $3, $4, $5, $6, $7, $8)
234 "#,
235 )
236 .bind(&id)
237 .bind(&request.function_name)
238 .bind(&input_json)
239 .bind(max_attempts)
240 .bind(&request.correlation_id.0)
241 .bind(serde_json::to_value(&request.actor).map_err(map_serde_error)?)
242 .bind(request.tenant_id.as_ref().map(|tenant| &tenant.0))
243 .bind(request.tenancy_mode.as_str())
244 .execute(&mut **tx)
245 .await
246 .map_err(map_runtime_error)
247 }
248 .instrument(span)
249 .await?;
250
251 Ok(EnqueuedFunctionRun {
252 id,
253 function_name: request.function_name,
254 correlation_id: request.correlation_id.0,
255 trace: request.trace,
256 max_attempts,
257 })
258 }
259
260 pub(crate) async fn record_function_enqueued(&self, run: &EnqueuedFunctionRun) {
261 self.record_function_execution_log(
262 &FunctionLogContext {
263 id: run.id.clone(),
264 function_name: run.function_name.clone(),
265 correlation_id: run.correlation_id.clone(),
266 trace: run.trace.clone(),
267 },
268 ExecutionLogSeverity::Info,
269 "Function run enqueued",
270 serde_json::json!({
271 "attempt": 0,
272 "max_attempts": run.max_attempts,
273 }),
274 )
275 .await;
276 }
277
278 async fn record_function_execution_log(
279 &self,
280 run: &FunctionLogContext,
281 severity: ExecutionLogSeverity,
282 body: &'static str,
283 attributes: Value,
284 ) {
285 emit_function_lifecycle_event(run, severity, body, &attributes, None);
286 if let Err(error) = insert_execution_log_projection(
287 &self.pool,
288 function_log_record(run, severity, body, attributes),
289 )
290 .await
291 {
292 tracing::warn!(
293 error = ?error,
294 function_run_id = %run.id,
295 "failed to write function execution log"
296 );
297 }
298 }
299}
300
301#[derive(Debug, Clone)]
302pub(crate) struct EnqueuedFunctionRun {
303 pub id: String,
304 function_name: String,
305 correlation_id: String,
306 trace: TraceContext,
307 max_attempts: i32,
308}
309
310#[derive(Debug, Clone)]
311pub struct RuntimeWorker {
312 pool: DbPool,
313 registry: Arc<FunctionRegistry>,
314 worker_id: String,
315}
316
317impl RuntimeWorker {
318 pub fn new(
319 pool: DbPool,
320 registry: Arc<FunctionRegistry>,
321 worker_id: impl Into<String>,
322 ) -> Self {
323 Self {
324 pool,
325 registry,
326 worker_id: worker_id.into(),
327 }
328 }
329
330 pub async fn claim_batch(&self, batch_size: i64) -> AppResult<Vec<ClaimedFunctionRun>> {
331 let span = tracing::info_span!(
332 "function_claim_batch",
333 worker_id = %self.worker_id,
334 lenso.execution.kind = "function_claim",
335 lenso.execution.name = "function.claim_batch",
336 );
337
338 async {
339 let runs = sqlx::query_as::<_, FunctionRunRow>(
340 r#"
341 with claimed as (
342 select id
343 from runtime.function_runs
344 where (
345 status in ('pending', 'failed')
346 and available_at <= now()
347 )
348 or (
349 status = 'processing'
350 and locked_at <= now() - ($1::double precision * interval '1 second')
351 )
352 order by available_at asc, created_at asc
353 limit $2
354 for update skip locked
355 )
356 update runtime.function_runs function_run
357 set status = 'processing',
358 locked_at = now(),
359 locked_by = $3,
360 started_at = coalesce(started_at, now()),
361 last_error = null,
362 updated_at = now()
363 from claimed
364 where function_run.id = claimed.id
365 returning
366 function_run.id,
367 function_run.function_name,
368 function_run.input_json,
369 function_run.attempts,
370 function_run.max_attempts,
371 function_run.correlation_id,
372 function_run.actor,
373 function_run.tenant_id,
374 function_run.tenancy_mode
375 "#,
376 )
377 .bind(stale_processing_lock_seconds())
378 .bind(batch_size)
379 .bind(&self.worker_id)
380 .fetch_all(&self.pool)
381 .await
382 .map(|rows| {
383 rows.into_iter()
384 .map(TryInto::try_into)
385 .collect::<AppResult<Vec<_>>>()
386 })
387 .map_err(map_runtime_error)??;
388
389 for run in &runs {
390 self.record_function_execution_log(
391 run,
392 ExecutionLogSeverity::Info,
393 "Function run claimed",
394 serde_json::json!({
395 "attempt": run.attempts + 1,
396 "max_attempts": run.max_attempts,
397 "worker_id": self.worker_id,
398 }),
399 )
400 .await;
401 }
402
403 Ok(runs)
404 }
405 .instrument(span)
406 .await
407 }
408
409 pub async fn claim_and_run_batch(&self, batch_size: i64) -> AppResult<usize> {
410 let span = tracing::info_span!(
411 "function_worker_loop",
412 worker_id = %self.worker_id,
413 lenso.execution.kind = "worker_loop",
414 lenso.execution.name = "runtime_worker.claim_and_run_batch",
415 );
416
417 async {
418 let runs = self.claim_batch(batch_size).await?;
419 let count = runs.len();
420
421 for run in runs {
422 self.run_claimed(run).await?;
423 }
424
425 Ok(count)
426 }
427 .instrument(span)
428 .await
429 }
430
431 async fn run_claimed(&self, run: ClaimedFunctionRun) -> AppResult<()> {
432 let span = tracing::info_span!(
433 "function_run",
434 lenso.correlation_id = tracing::field::Empty,
435 lenso.story_id = tracing::field::Empty,
436 lenso.function_run_id = tracing::field::Empty,
437 lenso.execution.kind = tracing::field::Empty,
438 lenso.execution.name = tracing::field::Empty,
439 );
440 record_runtime_span_attributes(
441 &span,
442 &RuntimeSpanAttributes::function(
443 run.correlation_id.clone(),
444 run.id.clone(),
445 run.function_name.clone(),
446 ),
447 );
448
449 async {
450 self.record_function_execution_log(
451 &run,
452 ExecutionLogSeverity::Info,
453 "Function run started",
454 serde_json::json!({
455 "attempt": run.attempts + 1,
456 "max_attempts": run.max_attempts,
457 "worker_id": self.worker_id,
458 }),
459 )
460 .await;
461
462 let Some(definition) = self.registry.get(&run.function_name) else {
463 let error = AppError::new(
464 ErrorCode::Internal,
465 format!("Runtime function {} is not registered", run.function_name),
466 )
467 .retryable();
468 self.mark_failed(&run, &error, RetryPolicy::default().initial_delay)
469 .await?;
470 return Ok(());
471 };
472
473 let attempt = u32::try_from(run.attempts + 1).unwrap_or(u32::MAX);
474 let ctx = ExecutionContext {
475 execution_id: ExecutionId(run.id.clone()),
476 function_name: run.function_name.clone(),
477 attempt,
478 queue: definition.queue.clone(),
479 correlation_id: CorrelationId::new(run.correlation_id.clone()),
480 causation_id: run.causation_id.clone(),
481 actor: run.actor.clone(),
482 tenant_id: run.tenant_id.clone(),
483 trace: run.trace.clone(),
484 deadline: None::<DateTime<Utc>>,
485 };
486
487 let observability = definition.handler.observability();
488 let started_at = Utc::now();
489 let started = Instant::now();
490 let result = definition.handler.call(ctx, run.input_json.clone()).await;
491 let duration_ms = started.elapsed().as_millis().try_into().unwrap_or(i64::MAX);
492 if let Some(observability) = observability {
493 self.record_function_handler_operation_log(
494 &run,
495 observability,
496 started_at,
497 duration_ms,
498 result.as_ref().err(),
499 )
500 .await;
501 }
502
503 match result {
504 Ok(_output) => self.mark_completed(&run).await,
505 Err(error) => {
506 self.mark_failed(&run, &error, definition.retry_policy.initial_delay)
507 .await
508 }
509 }
510 }
511 .instrument(span)
512 .await
513 }
514
515 pub async fn mark_completed(&self, run: &ClaimedFunctionRun) -> AppResult<()> {
516 sqlx::query(
517 r#"
518 update runtime.function_runs
519 set status = 'completed',
520 completed_at = now(),
521 locked_at = null,
522 locked_by = null,
523 last_error = null,
524 updated_at = now()
525 where id = $1
526 "#,
527 )
528 .bind(&run.id)
529 .execute(&self.pool)
530 .await
531 .map(|_| ())
532 .map_err(map_runtime_error)?;
533
534 self.record_function_execution_log(
535 run,
536 ExecutionLogSeverity::Info,
537 "Function run completed",
538 serde_json::json!({
539 "attempt": run.attempts + 1,
540 "max_attempts": run.max_attempts,
541 "worker_id": self.worker_id,
542 }),
543 )
544 .await;
545
546 Ok(())
547 }
548
549 pub async fn mark_failed(
550 &self,
551 run: &ClaimedFunctionRun,
552 error: &AppError,
553 retry_delay: Duration,
554 ) -> AppResult<()> {
555 let next_attempt = run.attempts + 1;
556 let status = if next_attempt >= run.max_attempts {
557 FunctionRunStatus::Dead
558 } else if error.retryable {
559 FunctionRunStatus::Failed
560 } else {
561 FunctionRunStatus::Dead
562 };
563
564 let span = tracing::info_span!(
565 "function_run_fail",
566 lenso.correlation_id = tracing::field::Empty,
567 lenso.story_id = tracing::field::Empty,
568 lenso.function_run_id = tracing::field::Empty,
569 lenso.execution.kind = tracing::field::Empty,
570 lenso.execution.name = tracing::field::Empty,
571 );
572 record_runtime_span_attributes(
573 &span,
574 &RuntimeSpanAttributes::function(
575 run.correlation_id.clone(),
576 run.id.clone(),
577 run.function_name.clone(),
578 ),
579 );
580
581 async {
582 sqlx::query(
583 r#"
584 update runtime.function_runs
585 set status = $2,
586 attempts = attempts + 1,
587 available_at = case
588 when $2 = 'failed' then now() + ($4::double precision * interval '1 second')
589 else available_at
590 end,
591 locked_at = null,
592 locked_by = null,
593 last_error = $3,
594 updated_at = now()
595 where id = $1
596 "#,
597 )
598 .bind(&run.id)
599 .bind(status.as_str())
600 .bind(error.public_message.as_str())
601 .bind(retry_delay_seconds(retry_delay))
602 .execute(&self.pool)
603 .await
604 .map(|_| ())
605 .map_err(map_runtime_error)?;
606
607 self.record_function_execution_log(
608 run,
609 ExecutionLogSeverity::Error,
610 if status == FunctionRunStatus::Dead {
611 "Function run marked dead"
612 } else {
613 "Function run failed"
614 },
615 serde_json::json!({
616 "attempt": next_attempt,
617 "max_attempts": run.max_attempts,
618 "status": status.as_str(),
619 "retryable": error.retryable,
620 "error": error.public_message,
621 "worker_id": self.worker_id,
622 }),
623 )
624 .await;
625
626 Ok(())
627 }
628 .instrument(span)
629 .await
630 }
631
632 async fn record_function_execution_log(
633 &self,
634 run: &ClaimedFunctionRun,
635 severity: ExecutionLogSeverity,
636 body: &'static str,
637 attributes: Value,
638 ) {
639 emit_function_lifecycle_event(run, severity, body, &attributes, Some(&self.worker_id));
640 if let Err(error) = insert_execution_log_projection(
641 &self.pool,
642 function_log_record(run, severity, body, attributes),
643 )
644 .await
645 {
646 tracing::warn!(
647 error = ?error,
648 function_run_id = %run.id,
649 "failed to write function execution log"
650 );
651 }
652 }
653
654 async fn record_function_handler_operation_log(
655 &self,
656 run: &ClaimedFunctionRun,
657 observability: FunctionHandlerObservability,
658 started_at: DateTime<Utc>,
659 duration_ms: i64,
660 error: Option<&AppError>,
661 ) {
662 let body = if error.is_some() {
663 "Function handler operation failed"
664 } else {
665 "Function handler operation completed"
666 };
667 let severity = if error.is_some() {
668 ExecutionLogSeverity::Error
669 } else {
670 ExecutionLogSeverity::Info
671 };
672 let attributes = function_handler_operation_attributes(
673 run,
674 &self.worker_id,
675 observability,
676 duration_ms,
677 error,
678 );
679 emit_function_lifecycle_event(run, severity, body, &attributes, Some(&self.worker_id));
680 if let Err(error) = insert_execution_log_projection(
681 &self.pool,
682 function_log_record(run, severity, body, attributes).with_occurred_at(started_at),
683 )
684 .await
685 {
686 tracing::warn!(
687 error = ?error,
688 function_run_id = %run.id,
689 "failed to write function handler operation log"
690 );
691 }
692 }
693}
694
695#[derive(Debug)]
696struct FunctionLogContext {
697 id: String,
698 function_name: String,
699 correlation_id: String,
700 trace: TraceContext,
701}
702
703trait FunctionLogSource {
704 fn id(&self) -> String;
705 fn function_name(&self) -> String;
706 fn correlation_id(&self) -> String;
707 fn trace(&self) -> TraceContext;
708}
709
710impl FunctionLogSource for FunctionLogContext {
711 fn id(&self) -> String {
712 self.id.clone()
713 }
714
715 fn function_name(&self) -> String {
716 self.function_name.clone()
717 }
718
719 fn correlation_id(&self) -> String {
720 self.correlation_id.clone()
721 }
722
723 fn trace(&self) -> TraceContext {
724 self.trace.clone()
725 }
726}
727
728impl FunctionLogSource for ClaimedFunctionRun {
729 fn id(&self) -> String {
730 self.id.clone()
731 }
732
733 fn function_name(&self) -> String {
734 self.function_name.clone()
735 }
736
737 fn correlation_id(&self) -> String {
738 self.correlation_id.clone()
739 }
740
741 fn trace(&self) -> TraceContext {
742 self.trace.clone()
743 }
744}
745
746fn emit_function_lifecycle_event(
747 run: &impl FunctionLogSource,
748 severity: ExecutionLogSeverity,
749 body: &'static str,
750 attributes: &Value,
751 worker_id: Option<&str>,
752) {
753 match severity {
754 ExecutionLogSeverity::Error => {
755 tracing::error!(
756 function_run_id = %run.id(),
757 function_name = %run.function_name(),
758 correlation_id = %run.correlation_id(),
759 worker_id = worker_id.unwrap_or(""),
760 attributes = %attributes,
761 "{body}"
762 );
763 }
764 _ => {
765 tracing::info!(
766 function_run_id = %run.id(),
767 function_name = %run.function_name(),
768 correlation_id = %run.correlation_id(),
769 worker_id = worker_id.unwrap_or(""),
770 attributes = %attributes,
771 "{body}"
772 );
773 }
774 }
775}
776
777fn function_log_record(
778 run: &impl FunctionLogSource,
779 severity: ExecutionLogSeverity,
780 body: impl Into<String>,
781 attributes: Value,
782) -> ExecutionLogProjectionRecord {
783 ExecutionLogProjectionRecord::from_runtime_attrs(
784 RuntimeSpanAttributes::function(run.correlation_id(), run.id(), run.function_name()),
785 severity,
786 body,
787 )
788 .with_attributes(attributes)
789 .with_trace(run.trace())
790}
791
792#[derive(Debug, Clone)]
793struct ExecutionLogProjectionRecord {
794 correlation_id: String,
795 execution_id: String,
796 execution_type: String,
797 execution_name: String,
798 severity: ExecutionLogSeverity,
799 body: String,
800 attributes: Value,
801 trace: TraceContext,
802 service_name: String,
803 occurred_at: Option<DateTime<Utc>>,
804}
805
806impl ExecutionLogProjectionRecord {
807 fn from_runtime_attrs(
808 attrs: RuntimeSpanAttributes,
809 severity: ExecutionLogSeverity,
810 body: impl Into<String>,
811 ) -> Self {
812 let execution_id = attrs
813 .function_run_id
814 .clone()
815 .or(attrs.outbox_event_id)
816 .unwrap_or_else(|| attrs.story_id.clone());
817
818 Self {
819 correlation_id: attrs.correlation_id,
820 execution_id,
821 execution_type: attrs.execution_kind,
822 execution_name: attrs.execution_name,
823 severity,
824 body: body.into(),
825 attributes: Value::Object(Default::default()),
826 trace: TraceContext::default(),
827 service_name: "lenso".to_owned(),
828 occurred_at: None,
829 }
830 }
831
832 fn with_attributes(mut self, attributes: Value) -> Self {
833 self.attributes = attributes;
834 self
835 }
836
837 fn with_trace(mut self, trace: TraceContext) -> Self {
838 self.trace = trace;
839 self
840 }
841
842 fn with_occurred_at(mut self, occurred_at: DateTime<Utc>) -> Self {
843 self.occurred_at = Some(occurred_at);
844 self
845 }
846}
847
848async fn insert_execution_log_projection(
849 pool: &DbPool,
850 record: ExecutionLogProjectionRecord,
851) -> AppResult<String> {
852 let id = format!("elog_{}", Uuid::now_v7());
853 let occurred_at = record.occurred_at.unwrap_or_else(Utc::now);
854
855 sqlx::query(
856 r#"
857 insert into platform.execution_logs (
858 id,
859 correlation_id,
860 story_id,
861 execution_id,
862 execution_type,
863 execution_name,
864 occurred_at,
865 severity,
866 body,
867 attributes,
868 trace_id,
869 span_id,
870 service_name,
871 redacted_fields
872 )
873 values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
874 "#,
875 )
876 .bind(&id)
877 .bind(&record.correlation_id)
878 .bind(&record.correlation_id)
879 .bind(&record.execution_id)
880 .bind(&record.execution_type)
881 .bind(&record.execution_name)
882 .bind(occurred_at)
883 .bind(record.severity.as_str())
884 .bind(&record.body)
885 .bind(normalize_log_attributes(record.attributes))
886 .bind(&record.trace.trace_id)
887 .bind(&record.trace.span_id)
888 .bind(&record.service_name)
889 .bind(Vec::<String>::new())
890 .execute(pool)
891 .await
892 .map_err(map_runtime_error)?;
893
894 Ok(id)
895}
896
897fn function_handler_operation_attributes(
898 run: &ClaimedFunctionRun,
899 worker_id: &str,
900 observability: FunctionHandlerObservability,
901 duration_ms: i64,
902 error: Option<&AppError>,
903) -> Value {
904 let mut attributes = match observability.attributes {
905 Value::Object(attributes) => attributes,
906 other => serde_json::Map::from_iter([("value".to_owned(), other)]),
907 };
908 attributes.insert("source".to_owned(), Value::String(observability.source));
909 attributes.insert("attempt".to_owned(), serde_json::json!(run.attempts + 1));
910 attributes.insert(
911 "max_attempts".to_owned(),
912 serde_json::json!(run.max_attempts),
913 );
914 attributes.insert("duration_ms".to_owned(), serde_json::json!(duration_ms));
915 attributes.insert("success".to_owned(), serde_json::json!(error.is_none()));
916 attributes.insert("worker_id".to_owned(), serde_json::json!(worker_id));
917 attributes.insert(
918 "function_name".to_owned(),
919 serde_json::json!(run.function_name),
920 );
921 attributes.insert("request_id".to_owned(), serde_json::json!(run.id));
922 attributes.insert("trace_id".to_owned(), serde_json::json!(run.trace.trace_id));
923 attributes.insert("span_id".to_owned(), serde_json::json!(run.trace.span_id));
924
925 if let Some(error) = error {
926 attributes.insert(
927 "error_code".to_owned(),
928 serde_json::json!(error.code.as_str()),
929 );
930 attributes.insert("error".to_owned(), serde_json::json!(error.public_message));
931 attributes.insert("retryable".to_owned(), serde_json::json!(error.retryable));
932 attributes.insert("error_details".to_owned(), serde_json::json!(error.details));
933 }
934
935 Value::Object(attributes)
936}
937
938fn normalize_log_attributes(attributes: Value) -> Value {
939 match attributes {
940 Value::Object(_) => attributes,
941 other => serde_json::json!({ "value": other }),
942 }
943}
944
945fn stale_processing_lock_seconds() -> f64 {
946 STALE_PROCESSING_LOCK_SECONDS as f64
947}
948
949fn retry_delay_seconds(delay: Duration) -> f64 {
950 delay.as_secs_f64()
951}
952
953type FunctionRunRow = (
954 String,
955 String,
956 Value,
957 i32,
958 i32,
959 String,
960 Value,
961 Option<String>,
962 String,
963);
964
965impl TryFrom<FunctionRunRow> for ClaimedFunctionRun {
966 type Error = AppError;
967
968 fn try_from(row: FunctionRunRow) -> Result<Self, Self::Error> {
969 let (
970 id,
971 function_name,
972 input_json,
973 attempts,
974 max_attempts,
975 correlation_id,
976 actor,
977 tenant_id,
978 tenancy_mode,
979 ) = row;
980 let runtime_context = input_json.get("_lenso_runtime");
981 let trace = runtime_context
982 .map(trace_context_from_headers)
983 .unwrap_or_default();
984 let causation_id = runtime_context
985 .and_then(|context| context.get("causation_id"))
986 .and_then(Value::as_str)
987 .map(ToOwned::to_owned);
988 Ok(Self {
989 id,
990 function_name,
991 input_json,
992 attempts,
993 max_attempts,
994 correlation_id,
995 actor: serde_json::from_value(actor).map_err(map_serde_error)?,
996 tenant_id: tenant_id.map(TenantId),
997 tenancy_mode: serde_json::from_value(Value::String(tenancy_mode))
998 .map_err(map_serde_error)?,
999 trace,
1000 causation_id,
1001 })
1002 }
1003}
1004
1005fn attach_runtime_context_to_input(
1006 input_json: &mut Value,
1007 correlation_id: &CorrelationId,
1008 trace: &TraceContext,
1009 causation_id: Option<&str>,
1010) {
1011 let mut runtime_context = trace_headers(trace, correlation_id);
1012 if let Some(causation_id) = causation_id {
1013 runtime_context["causation_id"] = Value::String(causation_id.to_owned());
1014 }
1015
1016 match input_json {
1017 Value::Object(object) => {
1018 object.insert("_lenso_runtime".to_owned(), runtime_context);
1019 }
1020 other => {
1021 *other = serde_json::json!({
1022 "payload": other.clone(),
1023 "_lenso_runtime": runtime_context,
1024 });
1025 }
1026 }
1027}
1028
1029pub(crate) fn map_runtime_error(source: sqlx::Error) -> AppError {
1030 AppError::new(ErrorCode::Internal, "Runtime operation failed").with_source(source)
1031}
1032
1033fn map_serde_error(source: serde_json::Error) -> AppError {
1034 AppError::new(ErrorCode::Internal, "Runtime payload serialization failed").with_source(source)
1035}