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