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