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