obeli_sk_executor/
executor.rs

1use crate::worker::{
2    FatalError, RunFinished, Worker, WorkerContext, WorkerError, WorkerResult, WorkerResultOk,
3};
4use assert_matches::assert_matches;
5use chrono::{DateTime, Utc};
6use concepts::prefixed_ulid::{DeploymentId, RunId};
7use concepts::storage::{
8    AppendEventsToExecution, AppendRequest, AppendResponseToExecution, DbErrorGeneric,
9    DbErrorWrite, DbExecutor, DbPool, ExecutionLog, LockedExecution,
10};
11use concepts::time::{ClockFn, Sleep};
12use concepts::{
13    ComponentId, ComponentRetryConfig, ComponentType, FunctionMetadata, StrVariant,
14    SupportedFunctionReturnValue,
15};
16use concepts::{ExecutionFailureKind, JoinSetId};
17use concepts::{ExecutionId, FunctionFqn, prefixed_ulid::ExecutorId};
18use concepts::{
19    FinishedExecutionFailure,
20    storage::{ExecutionRequest, Version},
21};
22use std::{
23    sync::{
24        Arc,
25        atomic::{AtomicBool, Ordering},
26    },
27    time::Duration,
28};
29use tokio::task::{AbortHandle, JoinHandle};
30use tracing::{Instrument, Level, Span, debug, error, info, info_span, instrument, trace, warn};
31
32#[derive(Debug, Clone)]
33pub struct ExecConfig {
34    pub lock_expiry: Duration,
35    pub tick_sleep: Duration,
36    pub batch_size: u32,
37    pub component_id: ComponentId,
38    pub task_limiter_global: Option<Arc<tokio::sync::Semaphore>>,
39    pub task_limiter_local: Option<Arc<tokio::sync::Semaphore>>,
40    pub executor_id: ExecutorId,
41    pub retry_config: ComponentRetryConfig,
42    pub locking_strategy: LockingStrategy,
43}
44
45pub struct ExecTask {
46    worker: Arc<dyn Worker>,
47    pub config: ExecConfig,
48    clock_fn: Box<dyn ClockFn>, // Used for obtaining current time when the execution finishes.
49    db_pool: Arc<dyn DbPool>,
50    locking_strategy_holder: LockingStrategyHolder,
51    worker_count_tx: tokio::sync::watch::Sender<usize>,
52    executor_close_watcher: tokio::sync::watch::Receiver<bool>,
53}
54
55#[derive(derive_more::Debug, Default)]
56pub struct ExecutionProgress {
57    #[debug(skip)]
58    #[allow(dead_code)]
59    executions: Vec<(ExecutionId, JoinHandle<()>)>,
60}
61
62impl ExecutionProgress {
63    #[cfg(feature = "test")]
64    pub async fn wait_for_tasks(self) -> Vec<ExecutionId> {
65        let mut vec = Vec::new();
66        for (exe, join_handle) in self.executions {
67            vec.push(exe);
68            join_handle.await.unwrap();
69        }
70        vec
71    }
72}
73
74#[derive(Clone, Copy, Debug, PartialEq, Eq)]
75pub enum WorkerType {
76    Activity,
77    Workflow,
78}
79
80#[derive(derive_more::Debug)]
81pub struct ExecutorTaskHandle {
82    #[debug(skip)]
83    is_closing: Arc<AtomicBool>,
84    #[debug(skip)]
85    abort_handle: AbortHandle,
86    component_id: ComponentId,
87    executor_id: ExecutorId,
88    deployment_id: DeploymentId,
89    executor_closing_signal_sender: tokio::sync::watch::Sender<bool>,
90    /// Tracks the number of worker tasks currently in-flight.
91    worker_count_rx: tokio::sync::watch::Receiver<usize>,
92}
93
94impl ExecutorTaskHandle {
95    #[instrument(name = "executor.close", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id,
96        deployment_id = %self.deployment_id))]
97    pub async fn close(&self) {
98        trace!("Gracefully closing");
99        self.is_closing.store(true, Ordering::Relaxed);
100        while !self.abort_handle.is_finished() {
101            tokio::time::sleep(Duration::from_millis(1)).await;
102        }
103        trace!("Signaling workflow tasks to unlock");
104        let _ = self.executor_closing_signal_sender.send(true);
105        let mut worker_count_rx = self.worker_count_rx.clone();
106        loop {
107            tokio::select! {
108                () = tokio::time::sleep(Duration::from_secs(5)) => {
109                    info!("Waiting for {} workers to shut down", *self.worker_count_rx.borrow());
110                }
111                _ = worker_count_rx.wait_for(|&count| count == 0) => {
112                    break;
113                }
114            }
115        }
116        debug!("Gracefully closed");
117    }
118
119    #[must_use]
120    pub fn component_id(&self) -> &ComponentId {
121        &self.component_id
122    }
123}
124
125impl Drop for ExecutorTaskHandle {
126    #[instrument(level = Level::DEBUG, name = "executor.drop", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id))]
127    fn drop(&mut self) {
128        if self.abort_handle.is_finished() {
129            return;
130        }
131        warn!("Aborting the executor task");
132        self.abort_handle.abort();
133    }
134}
135
136#[cfg(feature = "test")]
137pub fn extract_exported_ffqns_noext_test(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
138    extract_exported_ffqns_noext(worker)
139}
140
141fn extract_exported_ffqns_noext(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
142    worker
143        .exported_functions_noext()
144        .iter()
145        .map(|FunctionMetadata { ffqn, .. }| ffqn.clone())
146        .collect::<Arc<_>>()
147}
148
149#[derive(Debug, Clone, Copy)]
150pub enum LockingStrategy {
151    ByFfqns,
152    ByComponentDigest,
153}
154impl LockingStrategy {
155    fn holder(&self, ffqns: Arc<[FunctionFqn]>) -> LockingStrategyHolder {
156        match self {
157            LockingStrategy::ByFfqns => LockingStrategyHolder::ByFfqns(ffqns),
158            LockingStrategy::ByComponentDigest => LockingStrategyHolder::ByComponentId,
159        }
160    }
161}
162
163enum LockingStrategyHolder {
164    ByFfqns(Arc<[FunctionFqn]>),
165    ByComponentId,
166}
167
168#[derive(Default)]
169#[expect(dead_code)] // Stored permits limit semaphores until dropped.
170struct TaskLimiterPermit {
171    global: Option<tokio::sync::OwnedSemaphorePermit>,
172    local: Option<tokio::sync::OwnedSemaphorePermit>,
173}
174
175impl ExecTask {
176    #[cfg(feature = "test")]
177    pub fn new_test(
178        config: ExecConfig,
179        worker: Arc<dyn Worker>,
180        clock_fn: Box<dyn ClockFn>,
181        db_pool: Arc<dyn DbPool>,
182        ffqns: Arc<[FunctionFqn]>,
183    ) -> Self {
184        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
185        ExecTask {
186            worker,
187            locking_strategy_holder: config.locking_strategy.holder(ffqns),
188            config,
189            clock_fn,
190            db_pool,
191            worker_count_tx,
192            executor_close_watcher: tokio::sync::watch::channel(false).1,
193        }
194    }
195
196    #[cfg(feature = "test")]
197    pub fn new_all_ffqns_test(
198        worker: Arc<dyn Worker>,
199        config: ExecConfig,
200        clock_fn: Box<dyn ClockFn>,
201        db_pool: Arc<dyn DbPool>,
202    ) -> Self {
203        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
204        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
205        Self {
206            worker,
207            locking_strategy_holder: config.locking_strategy.holder(ffqns),
208            config,
209            clock_fn,
210            db_pool,
211            worker_count_tx,
212            executor_close_watcher: tokio::sync::watch::channel(false).1,
213        }
214    }
215
216    #[cfg(feature = "test")]
217    pub fn new_all_ffqns_test_with_close_watcher(
218        worker: Arc<dyn Worker>,
219        config: ExecConfig,
220        clock_fn: Box<dyn ClockFn>,
221        db_pool: Arc<dyn DbPool>,
222        executor_close_watcher: tokio::sync::watch::Receiver<bool>,
223    ) -> Self {
224        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
225        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
226        Self {
227            worker,
228            locking_strategy_holder: config.locking_strategy.holder(ffqns),
229            config,
230            clock_fn,
231            db_pool,
232            worker_count_tx,
233            executor_close_watcher,
234        }
235    }
236
237    pub fn spawn_new(
238        deployment_id: DeploymentId,
239        worker: Arc<dyn Worker>,
240        config: ExecConfig,
241        clock_fn: Box<dyn ClockFn>,
242        db_pool: Arc<dyn DbPool>,
243        sleep: impl Sleep + Clone + 'static,
244    ) -> ExecutorTaskHandle {
245        let is_closing = Arc::new(AtomicBool::default());
246        let is_closing_inner = is_closing.clone();
247        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
248        let component_id = config.component_id.clone();
249        let executor_id = config.executor_id;
250        let (worker_count_tx, worker_count_rx) = tokio::sync::watch::channel(0);
251        let (executor_closing_signal_sender, executor_close_watcher) =
252            tokio::sync::watch::channel(false);
253        let abort_handle = tokio::spawn(async move {
254            debug!(executor_id = %config.executor_id, component_id = %config.component_id, "Spawned executor");
255            let lock_strategy_holder = config.locking_strategy.holder(ffqns);
256            let task = ExecTask {
257                worker,
258                config,
259                db_pool,
260                locking_strategy_holder: lock_strategy_holder,
261                clock_fn: clock_fn.clone_box(),
262                worker_count_tx,
263                executor_close_watcher,
264            };
265            let mut old_err = None;
266            while !is_closing_inner.load(Ordering::Relaxed) {
267                let res = task.db_pool.db_exec_conn().await;
268                let res = log_err_if_new(res, &mut old_err);
269                if let Ok(db_exec) = res {
270                    let _ = task.tick(db_exec.as_ref(), clock_fn.now(), RunId::generate(), deployment_id).await;
271                    db_exec
272                        .wait_for_pending_by_component_digest(clock_fn.now(), &task.config.component_id.component_digest, {
273                            let sleep = sleep.clone();
274                            Box::pin(async move { sleep.sleep(task.config.tick_sleep).await })})
275                        .await;
276                } else  {
277                    sleep.sleep(task.config.tick_sleep).await;
278                }
279            }
280        })
281        .abort_handle();
282        ExecutorTaskHandle {
283            is_closing,
284            abort_handle,
285            component_id,
286            executor_id,
287            deployment_id,
288            executor_closing_signal_sender,
289            worker_count_rx,
290        }
291    }
292
293    fn acquire_task_permits(&self) -> Vec<TaskLimiterPermit> {
294        let mut locks = Vec::with_capacity(
295            usize::try_from(self.config.batch_size).expect("16 bit systems are unsupported"),
296        );
297        for _ in 0..self.config.batch_size {
298            match (
299                &self.config.task_limiter_global,
300                &self.config.task_limiter_local,
301            ) {
302                (Some(global), Some(local)) => {
303                    if let Ok(global) = global.clone().try_acquire_owned()
304                        && let Ok(local) = local.clone().try_acquire_owned()
305                    {
306                        locks.push(TaskLimiterPermit {
307                            global: Some(global),
308                            local: Some(local),
309                        });
310                    } else {
311                        break;
312                    }
313                }
314                (Some(global), None) => {
315                    if let Ok(global) = global.clone().try_acquire_owned() {
316                        locks.push(TaskLimiterPermit {
317                            global: Some(global),
318                            local: None,
319                        });
320                    } else {
321                        break;
322                    }
323                }
324                (None, Some(local)) => {
325                    if let Ok(local) = local.clone().try_acquire_owned() {
326                        locks.push(TaskLimiterPermit {
327                            global: None,
328                            local: Some(local),
329                        });
330                    } else {
331                        break;
332                    }
333                }
334                (None, None) => {
335                    locks.push(TaskLimiterPermit::default());
336                }
337            }
338        }
339        locks
340    }
341
342    #[cfg(feature = "test")]
343    pub async fn tick_test(&self, executed_at: DateTime<Utc>, run_id: RunId) -> ExecutionProgress {
344        use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
345
346        let db_exec = self.db_pool.db_exec_conn().await.unwrap();
347        self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
348            .await
349            .unwrap()
350    }
351
352    #[cfg(feature = "test")]
353    pub async fn tick_test_await(
354        &self,
355        executed_at: DateTime<Utc>,
356        run_id: RunId,
357    ) -> Vec<ExecutionId> {
358        use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
359
360        let db_exec = self.db_pool.db_exec_conn().await.unwrap();
361        self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
362            .await
363            .unwrap()
364            .wait_for_tasks()
365            .await
366    }
367
368    #[instrument(level = Level::TRACE, name = "executor.tick" skip_all, fields(executor_id = %self.config.executor_id, component_id = %self.config.component_id))]
369    async fn tick(
370        &self,
371        db_exec: &dyn DbExecutor,
372        executed_at: DateTime<Utc>,
373        run_id: RunId,
374        deployment_id: DeploymentId,
375    ) -> Result<ExecutionProgress, DbErrorWrite> {
376        let locked_executions = {
377            let mut permits = self.acquire_task_permits();
378            if permits.is_empty() {
379                return Ok(ExecutionProgress::default());
380            }
381            let lock_expires_at = executed_at + self.config.lock_expiry;
382            let batch_size = u32::try_from(permits.len()).expect("ExecConfig.batch_size is u32");
383            let locked_executions = match &self.locking_strategy_holder {
384                LockingStrategyHolder::ByFfqns(ffqns) => {
385                    db_exec
386                        .lock_pending_by_ffqns(
387                            batch_size,
388                            executed_at, // fetch expiring before now
389                            ffqns.clone(),
390                            executed_at, // created at
391                            self.config.component_id.clone(),
392                            deployment_id,
393                            self.config.executor_id,
394                            lock_expires_at,
395                            run_id,
396                            self.config.retry_config,
397                        )
398                        .await?
399                }
400                LockingStrategyHolder::ByComponentId => {
401                    db_exec
402                        .lock_pending_by_component_digest(
403                            batch_size,
404                            executed_at, // pending_at_or_sooner
405                            &self.config.component_id,
406                            deployment_id,
407                            executed_at, // created at
408                            self.config.executor_id,
409                            lock_expires_at,
410                            run_id,
411                            self.config.retry_config,
412                        )
413                        .await?
414                }
415            };
416            // Drop permits if too many were allocated.
417            while permits.len() > locked_executions.len() {
418                permits.pop();
419            }
420            assert_eq!(permits.len(), locked_executions.len());
421            locked_executions.into_iter().zip(permits)
422        };
423
424        let mut executions = Vec::with_capacity(locked_executions.len());
425        for (locked_execution, permit) in locked_executions {
426            let execution_id = locked_execution.execution_id.clone();
427            let join_handle = {
428                let worker = self.worker.clone();
429                let db_pool = self.db_pool.clone();
430                let clock_fn = self.clock_fn.clone_box();
431                let worker_span = info_span!(parent: None, "worker",
432                    "otel.name" = format!("worker {}", locked_execution.ffqn),
433                    %execution_id, %run_id,
434                    ffqn = %locked_execution.ffqn,
435                    executor_id = %self.config.executor_id,
436                    component_id = %self.config.component_id,
437                    %deployment_id,
438                );
439                locked_execution.metadata.enrich(&worker_span);
440                let component_type = self.config.component_id.component_type;
441                let worker_count_tx = self.worker_count_tx.clone();
442                worker_count_tx.send_modify(|n| *n += 1);
443                let executor_close_watcher = self.executor_close_watcher.clone();
444                tokio::spawn({
445                    let worker_span2 = worker_span.clone();
446                    let retry_config = self.config.retry_config;
447                    async move {
448                        let _permit = permit;
449                        let res = Self::run_worker(
450                            component_type,
451                            worker,
452                            db_pool.as_ref(),
453                            clock_fn,
454                            locked_execution,
455                            retry_config,
456                            worker_span2,
457                            executor_close_watcher
458                        )
459                        .await;
460                        if let Err(db_error) = res {
461                            error!("Got db error `{db_error:?}`, expecting watcher to mark execution as timed out");
462                        }
463                        worker_count_tx.send_modify(|n| *n -= 1);
464                    }
465                    .instrument(worker_span)
466                })
467            };
468            executions.push((execution_id, join_handle));
469        }
470        Ok(ExecutionProgress { executions })
471    }
472
473    #[expect(clippy::too_many_arguments)]
474    async fn run_worker(
475        component_type: ComponentType,
476        worker: Arc<dyn Worker>,
477        db_pool: &dyn DbPool,
478        clock_fn: Box<dyn ClockFn>,
479        locked_execution: LockedExecution,
480        retry_config: ComponentRetryConfig,
481        worker_span: Span,
482        executor_close_watcher: tokio::sync::watch::Receiver<bool>,
483    ) -> Result<(), DbErrorWrite> {
484        debug!("Worker::run starting");
485        trace!(
486            version = %locked_execution.next_version,
487            params = ?locked_execution.params,
488            event_history = ?locked_execution.event_history,
489            "Worker::run starting"
490        );
491        let can_be_retried = ExecutionLog::can_be_retried_after(
492            locked_execution.intermittent_event_count + 1,
493            retry_config.max_retries,
494            retry_config.retry_exp_backoff,
495        );
496        let unlock_expiry_on_limit_reached =
497            ExecutionLog::compute_retry_duration_when_retrying_forever(
498                locked_execution.intermittent_event_count + 1,
499                retry_config.retry_exp_backoff,
500            );
501        let ctx = WorkerContext {
502            execution_id: locked_execution.execution_id.clone(),
503            metadata: locked_execution.metadata,
504            ffqn: locked_execution.ffqn,
505            params: locked_execution.params,
506            event_history: locked_execution.event_history,
507            responses: locked_execution.responses,
508            version: locked_execution.next_version,
509            can_be_retried: can_be_retried.is_some(),
510            locked_event: locked_execution.locked_event,
511            worker_span,
512            executor_close_watcher,
513        };
514        let worker_result = worker.run(ctx).await;
515        debug!("Worker::run finished {worker_result:?}");
516        let result_obtained_at = clock_fn.now();
517        match Self::worker_result_to_execution_event(
518            component_type,
519            locked_execution.execution_id,
520            worker_result,
521            result_obtained_at,
522            locked_execution.parent,
523            can_be_retried,
524            unlock_expiry_on_limit_reached,
525        )? {
526            Some(append) => {
527                trace!("Appending {append:?}");
528                let db_exec = db_pool.db_exec_conn().await?;
529                append.append(db_exec.as_ref()).await
530            }
531            None => Ok(()),
532        }
533    }
534
535    /// Map the `WorkerError` to an optional append event
536    fn worker_result_to_execution_event(
537        component_type: ComponentType,
538        execution_id: ExecutionId,
539        worker_result: WorkerResult,
540        result_obtained_at: DateTime<Utc>,
541        parent: Option<(ExecutionId, JoinSetId)>,
542        can_be_retried: Option<Duration>,
543        unlock_expiry_on_limit_reached: Duration,
544    ) -> Result<Option<Append>, DbErrorWrite> {
545        Ok(match worker_result {
546            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
547                retval: ref retval @ SupportedFunctionReturnValue::Err(ref result_err),
548                version,
549                http_client_traces,
550            })) if component_type == ComponentType::Activity
551                && can_be_retried.is_some()
552                && !retval.is_permanent_variant() =>
553            {
554                // Interpret returned `err` variant as a retry request, unless it is a permanent variant
555                let detail = serde_json::to_string(result_err)
556                    .expect("SupportedFunctionReturnValue should be serializable to JSON");
557                let duration = can_be_retried.expect(
558                    "ActivityReturnedError must not be returned when retries are exhausted",
559                );
560                let expires_at = result_obtained_at + duration;
561                debug!("Retrying ActivityReturnedError after {duration:?} at {expires_at}");
562                let primary_event = ExecutionRequest::TemporarilyFailed {
563                    backoff_expires_at: expires_at,
564                    reason: StrVariant::Static("activity returned error"),
565                    detail: Some(detail),
566                    http_client_traces,
567                };
568                Some(Append {
569                    created_at: result_obtained_at,
570                    primary_event: AppendRequest {
571                        created_at: result_obtained_at,
572                        event: primary_event,
573                    },
574                    execution_id,
575                    version,
576                    child_finished: None,
577                })
578            }
579
580            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
581                retval: result,
582                version,
583                http_client_traces,
584            })) => {
585                info!("Execution finished: {result}");
586                let child_finished =
587                    parent.map(
588                        |(parent_execution_id, parent_join_set)| ChildFinishedResponse {
589                            parent_execution_id,
590                            parent_join_set,
591                            result: result.clone(),
592                        },
593                    );
594                let primary_event = AppendRequest {
595                    created_at: result_obtained_at,
596                    event: ExecutionRequest::Finished {
597                        retval: result,
598                        http_client_traces,
599                    },
600                };
601
602                Some(Append {
603                    created_at: result_obtained_at,
604                    primary_event,
605                    execution_id,
606                    version,
607                    child_finished,
608                })
609            }
610
611            WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher) => None,
612
613            WorkerResult::Err(err) => {
614                let reason_generic = err.to_string(); // Override with err's reason if no information is lost.
615
616                let (primary_event, child_finished, version) = match err {
617                    WorkerError::ExecutorClosing(version) => {
618                        let primary_event = ExecutionRequest::Unlocked {
619                            backoff_expires_at: result_obtained_at, // continue right when new executor starts
620                            reason: "executor closing".into(),
621                        };
622                        (primary_event, None, version)
623                    }
624                    WorkerError::TemporaryTimeout {
625                        http_client_traces,
626                        version,
627                    } => {
628                        if let Some(duration) = can_be_retried {
629                            let backoff_expires_at = result_obtained_at + duration;
630                            info!(
631                                "Temporary timeout, retrying after {duration:?} at {backoff_expires_at}"
632                            );
633                            (
634                                ExecutionRequest::TemporarilyTimedOut {
635                                    backoff_expires_at,
636                                    http_client_traces,
637                                },
638                                None,
639                                version,
640                            )
641                        } else {
642                            info!("Execution timed out");
643                            let result = SupportedFunctionReturnValue::ExecutionFailure(
644                                FinishedExecutionFailure {
645                                    kind: ExecutionFailureKind::TimedOut,
646                                    reason: None,
647                                    detail: None,
648                                },
649                            );
650                            let child_finished =
651                                parent.map(|(parent_execution_id, parent_join_set)| {
652                                    ChildFinishedResponse {
653                                        parent_execution_id,
654                                        parent_join_set,
655                                        result: result.clone(),
656                                    }
657                                });
658                            (
659                                ExecutionRequest::Finished {
660                                    retval: result,
661                                    http_client_traces,
662                                },
663                                child_finished,
664                                version,
665                            )
666                        }
667                    }
668                    WorkerError::DbError(db_error) => {
669                        return Err(db_error);
670                    }
671                    WorkerError::ActivityTrap {
672                        reason: _, // reason_generic contains trap_kind + reason
673                        trap_kind,
674                        detail,
675                        version,
676                        http_client_traces,
677                    } => {
678                        if let Some(duration) = can_be_retried {
679                            let expires_at = result_obtained_at + duration;
680                            debug!(
681                                "Retrying activity with `{trap_kind}` execution after {duration:?} at {expires_at}"
682                            );
683                            (
684                                ExecutionRequest::TemporarilyFailed {
685                                    reason: StrVariant::from(reason_generic),
686                                    backoff_expires_at: expires_at,
687                                    detail,
688                                    http_client_traces,
689                                },
690                                None,
691                                version,
692                            )
693                        } else {
694                            info!(
695                                "Activity with `{trap_kind}` marked as permanent failure - {reason_generic}"
696                            );
697                            let result = SupportedFunctionReturnValue::ExecutionFailure(
698                                FinishedExecutionFailure {
699                                    reason: Some(reason_generic),
700                                    kind: ExecutionFailureKind::Uncategorized,
701                                    detail,
702                                },
703                            );
704                            let child_finished =
705                                parent.map(|(parent_execution_id, parent_join_set)| {
706                                    ChildFinishedResponse {
707                                        parent_execution_id,
708                                        parent_join_set,
709                                        result: result.clone(),
710                                    }
711                                });
712                            (
713                                ExecutionRequest::Finished {
714                                    retval: result,
715                                    http_client_traces,
716                                },
717                                child_finished,
718                                version,
719                            )
720                        }
721                    }
722                    WorkerError::LimitReached {
723                        reason,
724                        version: new_version,
725                    } => {
726                        let expires_at = result_obtained_at + unlock_expiry_on_limit_reached;
727                        warn!(
728                            "Limit reached: {reason}, unlocking after {unlock_expiry_on_limit_reached:?} at {expires_at}"
729                        );
730                        (
731                            ExecutionRequest::Unlocked {
732                                backoff_expires_at: expires_at,
733                                reason: StrVariant::from(reason),
734                            },
735                            None,
736                            new_version,
737                        )
738                    }
739                    WorkerError::FatalError(FatalError::Cancelled, _version) => {
740                        unreachable!(
741                            "activity workers must return DbUpdatedByWorkerOrWatcher, cancellation append happens in CancelRegistry::cancel_activity"
742                        )
743                    }
744                    WorkerError::FatalError(fatal_error, version) => {
745                        warn!("Fatal worker error - {fatal_error:?}");
746                        let result = SupportedFunctionReturnValue::ExecutionFailure(
747                            FinishedExecutionFailure::from(fatal_error),
748                        );
749                        let child_finished =
750                            parent.map(|(parent_execution_id, parent_join_set)| {
751                                ChildFinishedResponse {
752                                    parent_execution_id,
753                                    parent_join_set,
754                                    result: result.clone(),
755                                }
756                            });
757                        (
758                            ExecutionRequest::Finished {
759                                retval: result,
760                                http_client_traces: None,
761                            },
762                            child_finished,
763                            version,
764                        )
765                    }
766                };
767                Some(Append {
768                    created_at: result_obtained_at,
769                    primary_event: AppendRequest {
770                        created_at: result_obtained_at,
771                        event: primary_event,
772                    },
773                    execution_id,
774                    version,
775                    child_finished,
776                })
777            }
778        })
779    }
780}
781
782#[derive(Debug, Clone)]
783pub(crate) struct ChildFinishedResponse {
784    pub(crate) parent_execution_id: ExecutionId,
785    pub(crate) parent_join_set: JoinSetId,
786    pub(crate) result: SupportedFunctionReturnValue,
787}
788
789#[derive(Debug, Clone)]
790pub(crate) struct Append {
791    pub(crate) created_at: DateTime<Utc>,
792    pub(crate) primary_event: AppendRequest,
793    pub(crate) execution_id: ExecutionId,
794    pub(crate) version: Version,
795    pub(crate) child_finished: Option<ChildFinishedResponse>,
796}
797
798impl Append {
799    pub(crate) async fn append(self, db_exec: &dyn DbExecutor) -> Result<(), DbErrorWrite> {
800        if let Some(child_finished) = self.child_finished {
801            assert_matches!(
802                &self.primary_event,
803                AppendRequest {
804                    event: ExecutionRequest::Finished { .. },
805                    ..
806                }
807            );
808            let child_execution_id = assert_matches!(self.execution_id.clone(), ExecutionId::Derived(derived) => derived);
809            let events = AppendEventsToExecution {
810                execution_id: self.execution_id,
811                version: self.version.clone(),
812                batch: vec![self.primary_event],
813            };
814            let response = AppendResponseToExecution {
815                parent_execution_id: child_finished.parent_execution_id,
816                created_at: self.created_at,
817                join_set_id: child_finished.parent_join_set,
818                child_execution_id,
819                finished_version: self.version, // Since self.primary_event is a finished event, the version will remain the same.
820                result: child_finished.result,
821            };
822
823            db_exec
824                .append_batch_respond_to_parent(events, response, self.created_at)
825                .await?;
826        } else {
827            db_exec
828                .append(self.execution_id, self.version, self.primary_event)
829                .await?;
830        }
831        Ok(())
832    }
833}
834
835fn log_err_if_new<T>(
836    res: Result<T, DbErrorGeneric>,
837    old_err: &mut Option<DbErrorGeneric>,
838) -> Result<T, ()> {
839    match (res, &old_err) {
840        (Ok(ok), _) => {
841            *old_err = None;
842            Ok(ok)
843        }
844        (Err(err), Some(old)) if err == *old => Err(()),
845        (Err(err), _) => {
846            warn!("Tick failed: {err:?}");
847            *old_err = Some(err);
848            Err(())
849        }
850    }
851}
852
853#[cfg(any(test, feature = "test"))]
854pub mod simple_worker {
855    use crate::worker::{Worker, WorkerContext, WorkerResult};
856    use async_trait::async_trait;
857    use concepts::{
858        FunctionFqn, FunctionMetadata, ParameterTypes, RETURN_TYPE_DUMMY,
859        storage::{HistoryEvent, Version},
860    };
861    use indexmap::IndexMap;
862    use std::sync::Arc;
863    use tracing::trace;
864
865    pub(crate) const FFQN_SOME: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn");
866    pub type SimpleWorkerResultMap =
867        Arc<std::sync::Mutex<IndexMap<Version, (Vec<HistoryEvent>, WorkerResult)>>>;
868
869    #[derive(Clone, Debug)]
870    pub struct SimpleWorker {
871        pub worker_results_rev: SimpleWorkerResultMap,
872        pub ffqn: FunctionFqn,
873        exported: [FunctionMetadata; 1],
874    }
875
876    impl SimpleWorker {
877        #[must_use]
878        pub fn with_single_result(res: WorkerResult) -> Self {
879            Self::with_worker_results_rev(Arc::new(std::sync::Mutex::new(IndexMap::from([(
880                Version::new(2),
881                (vec![], res),
882            )]))))
883        }
884
885        #[must_use]
886        pub fn with_ffqn(self, ffqn: FunctionFqn) -> Self {
887            Self {
888                worker_results_rev: self.worker_results_rev,
889                exported: [FunctionMetadata {
890                    ffqn: ffqn.clone(),
891                    parameter_types: ParameterTypes::default(),
892                    return_type: RETURN_TYPE_DUMMY,
893                    extension: None,
894                    submittable: true,
895                }],
896                ffqn,
897            }
898        }
899
900        #[must_use]
901        pub fn with_worker_results_rev(worker_results_rev: SimpleWorkerResultMap) -> Self {
902            Self {
903                worker_results_rev,
904                ffqn: FFQN_SOME,
905                exported: [FunctionMetadata {
906                    ffqn: FFQN_SOME,
907                    parameter_types: ParameterTypes::default(),
908                    return_type: RETURN_TYPE_DUMMY,
909                    extension: None,
910                    submittable: true,
911                }],
912            }
913        }
914    }
915
916    #[async_trait]
917    impl Worker for SimpleWorker {
918        async fn run(&self, ctx: WorkerContext) -> WorkerResult {
919            let (expected_version, (expected_eh, worker_result)) =
920                self.worker_results_rev.lock().unwrap().pop().unwrap();
921            trace!(%expected_version, version = %ctx.version, ?expected_eh, eh = ?ctx.event_history, "Running SimpleWorker");
922            assert_eq!(expected_version, ctx.version);
923            assert_eq!(
924                expected_eh,
925                ctx.event_history
926                    .iter()
927                    .map(|(event, _version)| event.clone())
928                    .collect::<Vec<_>>()
929            );
930            worker_result
931        }
932
933        fn exported_functions_noext(&self) -> &[FunctionMetadata] {
934            &self.exported
935        }
936    }
937}
938
939#[cfg(test)]
940mod tests {
941    use self::simple_worker::SimpleWorker;
942    use super::*;
943    use crate::{expired_timers_watcher, worker::WorkerResult};
944    use assert_matches::assert_matches;
945    use async_trait::async_trait;
946    use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
947    use concepts::storage::{
948        CreateRequest, DbConnectionTest, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent,
949    };
950    use concepts::storage::{DbPoolCloseable, LockedBy};
951    use concepts::storage::{
952        ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState, PendingStatePendingAt,
953    };
954    use concepts::time::{ConstClock, Now};
955    use concepts::{
956        FunctionMetadata, JoinSetKind, ParameterTypes, Params, RETURN_TYPE_DUMMY,
957        SUPPORTED_RETURN_VALUE_OK_EMPTY, StrVariant, SupportedFunctionReturnValue, TrapKind,
958    };
959    use db_tests::Database;
960    use indexmap::IndexMap;
961    use rstest::rstest;
962    use simple_worker::FFQN_SOME;
963    use std::{fmt::Debug, future::Future, ops::Deref, sync::Arc};
964    use test_db_macro::expand_enum_database;
965    use test_utils::set_up;
966    use test_utils::sim_clock::SimClock;
967
968    pub(crate) const FFQN_CHILD: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn-child");
969
970    async fn tick_fn<W: Worker + Debug>(
971        config: ExecConfig,
972        clock_fn: Box<dyn ClockFn>,
973        db_pool: Arc<dyn DbPool>,
974        worker: Arc<W>,
975        executed_at: DateTime<Utc>,
976    ) -> Vec<ExecutionId> {
977        trace!("Ticking with {worker:?}");
978        let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
979        let executor = ExecTask::new_test(config, worker, clock_fn, db_pool, ffqns);
980        executor
981            .tick_test_await(executed_at, RunId::generate())
982            .await
983    }
984
985    #[expand_enum_database]
986    #[rstest]
987    #[tokio::test]
988    async fn execute_simple_lifecycle_tick_based(
989        database: Database,
990        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
991        locking_strategy: LockingStrategy,
992    ) {
993        set_up();
994        let created_at = Now.now();
995        let (_guard, db_pool, db_close) = database.set_up().await;
996        let db_connection = db_pool.connection_test().await.unwrap();
997        execute_simple_lifecycle_tick_based_inner(
998            db_connection.as_ref(),
999            db_pool.clone(),
1000            Box::new(ConstClock(created_at)),
1001            locking_strategy,
1002        )
1003        .await;
1004        drop(db_connection);
1005        db_close.close().await;
1006    }
1007
1008    async fn execute_simple_lifecycle_tick_based_inner(
1009        db_connection: &dyn DbConnectionTest,
1010        db_pool: Arc<dyn DbPool>,
1011        clock_fn: Box<dyn ClockFn>,
1012        locking_strategy: LockingStrategy,
1013    ) {
1014        let created_at = clock_fn.now();
1015        let exec_config = ExecConfig {
1016            batch_size: 1,
1017            lock_expiry: Duration::from_secs(1),
1018            tick_sleep: Duration::from_millis(100),
1019            component_id: ComponentId::dummy_activity(),
1020            task_limiter_global: None,
1021            task_limiter_local: None,
1022            executor_id: ExecutorId::generate(),
1023            retry_config: ComponentRetryConfig::ZERO,
1024            locking_strategy,
1025        };
1026
1027        let execution_log = create_and_tick(
1028            CreateAndTickConfig {
1029                execution_id: ExecutionId::generate(),
1030                created_at,
1031                executed_at: created_at,
1032            },
1033            clock_fn,
1034            db_connection,
1035            db_pool,
1036            exec_config,
1037            Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1038                WorkerResultOk::RunFinished(RunFinished {
1039                    retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1040                    version: Version::new(2),
1041                    http_client_traces: None,
1042                }),
1043            ))),
1044            tick_fn,
1045        )
1046        .await;
1047        assert_matches!(
1048            execution_log.events.get(2).unwrap(),
1049            ExecutionEvent {
1050                event: ExecutionRequest::Finished {
1051                    retval: SupportedFunctionReturnValue::Ok(None),
1052                    http_client_traces: None
1053                },
1054                created_at: _,
1055                backtrace_id: None,
1056                version: Version(2),
1057            }
1058        );
1059    }
1060
1061    #[rstest]
1062    #[tokio::test]
1063    async fn execute_simple_lifecycle_task_based_mem(
1064        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1065        locking_strategy: LockingStrategy,
1066    ) {
1067        set_up();
1068        let created_at = Now.now();
1069        let clock_fn = Box::new(ConstClock(created_at));
1070        let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1071        let exec_config = ExecConfig {
1072            batch_size: 1,
1073            lock_expiry: Duration::from_secs(1),
1074            tick_sleep: Duration::ZERO,
1075            component_id: ComponentId::dummy_activity(),
1076            task_limiter_global: None,
1077            task_limiter_local: None,
1078            executor_id: ExecutorId::generate(),
1079            retry_config: ComponentRetryConfig::ZERO,
1080            locking_strategy,
1081        };
1082
1083        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1084            WorkerResultOk::RunFinished(RunFinished {
1085                retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1086                version: Version::new(2),
1087                http_client_traces: None,
1088            }),
1089        )));
1090        let db_connection = db_pool.connection_test().await.unwrap();
1091
1092        let execution_log = create_and_tick(
1093            CreateAndTickConfig {
1094                execution_id: ExecutionId::generate(),
1095                created_at,
1096                executed_at: created_at,
1097            },
1098            clock_fn,
1099            db_connection.as_ref(),
1100            db_pool,
1101            exec_config,
1102            worker,
1103            tick_fn,
1104        )
1105        .await;
1106        assert_matches!(
1107            execution_log.events.get(2).unwrap(),
1108            ExecutionEvent {
1109                event: ExecutionRequest::Finished {
1110                    retval: SupportedFunctionReturnValue::Ok(None),
1111                    http_client_traces: None
1112                },
1113                created_at: _,
1114                backtrace_id: None,
1115                version: Version(2),
1116            }
1117        );
1118        db_close.close().await;
1119    }
1120
1121    struct CreateAndTickConfig {
1122        execution_id: ExecutionId,
1123        created_at: DateTime<Utc>,
1124        executed_at: DateTime<Utc>,
1125    }
1126
1127    async fn create_and_tick<
1128        W: Worker,
1129        T: FnMut(ExecConfig, Box<dyn ClockFn>, Arc<dyn DbPool>, Arc<W>, DateTime<Utc>) -> F,
1130        F: Future<Output = Vec<ExecutionId>>,
1131    >(
1132        config: CreateAndTickConfig,
1133        clock_fn: Box<dyn ClockFn>,
1134        db_connection: &dyn DbConnectionTest,
1135        db_pool: Arc<dyn DbPool>,
1136        exec_config: ExecConfig,
1137        worker: Arc<W>,
1138        mut tick: T,
1139    ) -> ExecutionLog {
1140        // Create an execution
1141        db_connection
1142            .create(CreateRequest {
1143                created_at: config.created_at,
1144                execution_id: config.execution_id.clone(),
1145                ffqn: FFQN_SOME,
1146                params: Params::empty(),
1147                parent: None,
1148                metadata: concepts::ExecutionMetadata::empty(),
1149                scheduled_at: config.created_at,
1150                component_id: ComponentId::dummy_activity(),
1151                deployment_id: DEPLOYMENT_ID_DUMMY,
1152                scheduled_by: None,
1153                paused: false,
1154            })
1155            .await
1156            .unwrap();
1157        // execute!
1158        tick(exec_config, clock_fn, db_pool, worker, config.executed_at).await;
1159        let execution_log = db_connection.get(&config.execution_id).await.unwrap();
1160        debug!("Execution history after tick: {execution_log:?}");
1161        // check that DB contains Created and Locked events.
1162        let actually_created_at = assert_matches!(
1163            execution_log.events.first().unwrap(),
1164            ExecutionEvent {
1165                event: ExecutionRequest::Created { .. },
1166                created_at: actually_created_at,
1167                backtrace_id: None,
1168                version: Version(0),
1169            }
1170            => *actually_created_at
1171        );
1172        assert_eq!(config.created_at, actually_created_at);
1173        let locked_at = assert_matches!(
1174            execution_log.events.get(1).unwrap(),
1175            ExecutionEvent {
1176                event: ExecutionRequest::Locked { .. },
1177                created_at: locked_at,
1178                backtrace_id: None,
1179                version: Version(1),
1180            } if config.created_at <= *locked_at
1181            => *locked_at
1182        );
1183        assert_matches!(execution_log.events.get(2).unwrap(), ExecutionEvent {
1184            event: _,
1185            created_at: executed_at,
1186            backtrace_id: None,
1187            version: Version(2),
1188        } if *executed_at >= locked_at);
1189        execution_log
1190    }
1191
1192    #[rstest]
1193    #[tokio::test]
1194    async fn activity_trap_should_trigger_an_execution_retry(
1195        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1196        locking_strategy: LockingStrategy,
1197    ) {
1198        set_up();
1199        let sim_clock = SimClock::default();
1200        let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1201        let retry_exp_backoff = Duration::from_millis(100);
1202        let retry_config = ComponentRetryConfig {
1203            max_retries: Some(1),
1204            retry_exp_backoff,
1205        };
1206        let exec_config = ExecConfig {
1207            batch_size: 1,
1208            lock_expiry: Duration::from_secs(1),
1209            tick_sleep: Duration::ZERO,
1210            component_id: ComponentId::dummy_activity(),
1211            task_limiter_global: None,
1212            task_limiter_local: None,
1213            executor_id: ExecutorId::generate(),
1214            retry_config,
1215            locking_strategy,
1216        };
1217        let expected_reason = "error reason";
1218        let expected_detail = "error detail";
1219        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
1220            WorkerError::ActivityTrap {
1221                reason: expected_reason.to_string(),
1222                trap_kind: concepts::TrapKind::Trap,
1223                detail: Some(expected_detail.to_string()),
1224                version: Version::new(2),
1225                http_client_traces: None,
1226            },
1227        )));
1228        debug!(now = %sim_clock.now(), "Creating an execution that should fail");
1229        let db_connection = db_pool.connection_test().await.unwrap();
1230        let execution_log = create_and_tick(
1231            CreateAndTickConfig {
1232                execution_id: ExecutionId::generate(),
1233                created_at: sim_clock.now(),
1234                executed_at: sim_clock.now(),
1235            },
1236            sim_clock.clone_box(),
1237            db_connection.as_ref(),
1238            db_pool.clone(),
1239            exec_config.clone(),
1240            worker,
1241            tick_fn,
1242        )
1243        .await;
1244        assert_eq!(3, execution_log.events.len());
1245        {
1246            let (reason, detail, at, expires_at) = assert_matches!(
1247                &execution_log.events.get(2).unwrap(),
1248                ExecutionEvent {
1249                    event: ExecutionRequest::TemporarilyFailed {
1250                        reason,
1251                        detail,
1252                        backoff_expires_at,
1253                        http_client_traces: None,
1254                    },
1255                    created_at: at,
1256                    backtrace_id: None,
1257                    version: Version(2),
1258                }
1259                => (reason, detail, *at, *backoff_expires_at)
1260            );
1261            assert_eq!(format!("activity trap: {expected_reason}"), reason.deref());
1262            assert_eq!(Some(expected_detail), detail.as_deref());
1263            assert_eq!(at, sim_clock.now());
1264            assert_eq!(sim_clock.now() + retry_config.retry_exp_backoff, expires_at);
1265        }
1266        let worker = Arc::new(SimpleWorker::with_worker_results_rev(Arc::new(
1267            std::sync::Mutex::new(IndexMap::from([(
1268                Version::new(4),
1269                (
1270                    vec![],
1271                    WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
1272                        retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1273                        version: Version::new(4),
1274                        http_client_traces: None,
1275                    })),
1276                ),
1277            )])),
1278        )));
1279        // noop until `retry_exp_backoff` expires
1280        assert!(
1281            tick_fn(
1282                exec_config.clone(),
1283                sim_clock.clone_box(),
1284                db_pool.clone(),
1285                worker.clone(),
1286                sim_clock.now(),
1287            )
1288            .await
1289            .is_empty()
1290        );
1291        // tick again to finish the execution
1292        sim_clock.move_time_forward(retry_config.retry_exp_backoff);
1293        tick_fn(
1294            exec_config,
1295            sim_clock.clone_box(),
1296            db_pool.clone(),
1297            worker,
1298            sim_clock.now(),
1299        )
1300        .await;
1301        let execution_log = {
1302            let db_connection = db_pool.connection_test().await.unwrap();
1303            db_connection
1304                .get(&execution_log.execution_id)
1305                .await
1306                .unwrap()
1307        };
1308        debug!(now = %sim_clock.now(), "Execution history after second tick: {execution_log:?}");
1309        assert_matches!(
1310            execution_log.events.get(3).unwrap(),
1311            ExecutionEvent {
1312                event: ExecutionRequest::Locked { .. },
1313                created_at: at,
1314                backtrace_id: None,
1315                version: Version(3),
1316            } if *at == sim_clock.now()
1317        );
1318        assert_matches!(
1319            execution_log.events.get(4).unwrap(),
1320            ExecutionEvent {
1321                event: ExecutionRequest::Finished {
1322                    retval: SupportedFunctionReturnValue::Ok(None),
1323                    http_client_traces: None
1324                },
1325                created_at: finished_at,
1326                backtrace_id: None,
1327                version: Version(4),
1328            } if *finished_at == sim_clock.now()
1329        );
1330        db_close.close().await;
1331    }
1332
1333    #[rstest]
1334    #[tokio::test]
1335    async fn activity_trap_should_not_be_retried_if_no_retries_are_set(
1336        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1337        locking_strategy: LockingStrategy,
1338    ) {
1339        set_up();
1340        let created_at = Now.now();
1341        let clock_fn = Box::new(ConstClock(created_at));
1342        let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1343        let exec_config = ExecConfig {
1344            batch_size: 1,
1345            lock_expiry: Duration::from_secs(1),
1346            tick_sleep: Duration::ZERO,
1347            component_id: ComponentId::dummy_activity(),
1348            task_limiter_global: None,
1349            task_limiter_local: None,
1350            executor_id: ExecutorId::generate(),
1351            retry_config: ComponentRetryConfig::ZERO,
1352            locking_strategy,
1353        };
1354
1355        let reason = "error reason";
1356        let expected_reason = format!("activity trap: {reason}");
1357        let expected_detail = "error detail";
1358        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
1359            WorkerError::ActivityTrap {
1360                reason: reason.to_string(),
1361                trap_kind: concepts::TrapKind::Trap,
1362                detail: Some(expected_detail.to_string()),
1363                version: Version::new(2),
1364                http_client_traces: None,
1365            },
1366        )));
1367        let execution_log = create_and_tick(
1368            CreateAndTickConfig {
1369                execution_id: ExecutionId::generate(),
1370                created_at,
1371                executed_at: created_at,
1372            },
1373            clock_fn,
1374            db_pool.connection_test().await.unwrap().as_ref(),
1375            db_pool.clone(),
1376            exec_config.clone(),
1377            worker,
1378            tick_fn,
1379        )
1380        .await;
1381        assert_eq!(3, execution_log.events.len());
1382        let (reason, kind, detail) = assert_matches!(
1383            &execution_log.events.get(2).unwrap(),
1384            ExecutionEvent {
1385                event: ExecutionRequest::Finished{
1386                    retval: SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure{reason, kind, detail}),
1387                    http_client_traces: None
1388                },
1389                created_at: at,
1390                backtrace_id: None,
1391                version: Version(2),
1392            } if *at == created_at
1393            => (reason, kind, detail)
1394        );
1395
1396        assert_eq!(Some(expected_reason), *reason);
1397        assert_eq!(Some(expected_detail), detail.as_deref());
1398        assert_eq!(ExecutionFailureKind::Uncategorized, *kind);
1399
1400        db_close.close().await;
1401    }
1402
1403    #[rstest]
1404    #[tokio::test]
1405    async fn child_execution_permanently_failed_should_notify_parent_permanent_failure(
1406        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1407        locking_strategy: LockingStrategy,
1408    ) {
1409        let worker_error = WorkerError::ActivityTrap {
1410            reason: "error reason".to_string(),
1411            trap_kind: TrapKind::Trap,
1412            detail: Some("detail".to_string()),
1413            version: Version::new(2),
1414            http_client_traces: None,
1415        };
1416        let expected_child_err = FinishedExecutionFailure {
1417            kind: ExecutionFailureKind::Uncategorized,
1418            reason: Some("activity trap: error reason".to_string()),
1419            detail: Some("detail".to_string()),
1420        };
1421        child_execution_permanently_failed_should_notify_parent(
1422            WorkerResult::Err(worker_error),
1423            expected_child_err,
1424            locking_strategy,
1425        )
1426        .await;
1427    }
1428
1429    #[rstest]
1430    #[tokio::test]
1431    async fn child_execution_permanently_failed_handled_by_watcher_should_notify_parent_timeout(
1432        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1433        locking_strategy: LockingStrategy,
1434    ) {
1435        let expected_child_err = FinishedExecutionFailure {
1436            kind: ExecutionFailureKind::TimedOut,
1437            reason: None,
1438            detail: None,
1439        };
1440        child_execution_permanently_failed_should_notify_parent(
1441            WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher),
1442            expected_child_err,
1443            locking_strategy,
1444        )
1445        .await;
1446    }
1447
1448    async fn child_execution_permanently_failed_should_notify_parent(
1449        worker_result: WorkerResult,
1450        expected_child_err: FinishedExecutionFailure,
1451        locking_strategy: LockingStrategy,
1452    ) {
1453        use concepts::storage::JoinSetResponseEventOuter;
1454        const LOCK_EXPIRY: Duration = Duration::from_secs(1);
1455
1456        set_up();
1457        let sim_clock = SimClock::default();
1458        let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1459
1460        let parent_worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1461            WorkerResultOk::DbUpdatedByWorkerOrWatcher,
1462        )));
1463        let parent_execution_id = ExecutionId::generate();
1464        db_pool
1465            .connection()
1466            .await
1467            .unwrap()
1468            .create(CreateRequest {
1469                created_at: sim_clock.now(),
1470                execution_id: parent_execution_id.clone(),
1471                ffqn: FFQN_SOME,
1472                params: Params::empty(),
1473                parent: None,
1474                metadata: concepts::ExecutionMetadata::empty(),
1475                scheduled_at: sim_clock.now(),
1476                component_id: ComponentId::dummy_activity(),
1477                deployment_id: DEPLOYMENT_ID_DUMMY,
1478                scheduled_by: None,
1479                paused: false,
1480            })
1481            .await
1482            .unwrap();
1483        let parent_executor_id = ExecutorId::generate();
1484        tick_fn(
1485            ExecConfig {
1486                batch_size: 1,
1487                lock_expiry: LOCK_EXPIRY,
1488                tick_sleep: Duration::ZERO,
1489                component_id: ComponentId::dummy_activity(),
1490                task_limiter_global: None,
1491                task_limiter_local: None,
1492                executor_id: parent_executor_id,
1493                retry_config: ComponentRetryConfig::ZERO,
1494                locking_strategy,
1495            },
1496            sim_clock.clone_box(),
1497            db_pool.clone(),
1498            parent_worker,
1499            sim_clock.now(),
1500        )
1501        .await;
1502
1503        let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
1504        let child_execution_id = parent_execution_id.next_level(&join_set_id);
1505        // executor does not append anything, this should have been written by the worker:
1506        {
1507            let params = Params::empty();
1508            let child = CreateRequest {
1509                created_at: sim_clock.now(),
1510                execution_id: ExecutionId::Derived(child_execution_id.clone()),
1511                ffqn: FFQN_CHILD,
1512                params: params.clone(),
1513                parent: Some((parent_execution_id.clone(), join_set_id.clone())),
1514                metadata: concepts::ExecutionMetadata::empty(),
1515                scheduled_at: sim_clock.now(),
1516                component_id: ComponentId::dummy_activity(),
1517                deployment_id: DEPLOYMENT_ID_DUMMY,
1518                scheduled_by: None,
1519                paused: false,
1520            };
1521            let current_time = sim_clock.now();
1522            let join_set = AppendRequest {
1523                created_at: current_time,
1524                event: ExecutionRequest::HistoryEvent {
1525                    event: HistoryEvent::JoinSetCreate {
1526                        join_set_id: join_set_id.clone(),
1527                    },
1528                },
1529            };
1530            let child_exec_req = AppendRequest {
1531                created_at: current_time,
1532                event: ExecutionRequest::HistoryEvent {
1533                    event: HistoryEvent::JoinSetRequest {
1534                        join_set_id: join_set_id.clone(),
1535                        request: JoinSetRequest::ChildExecutionRequest {
1536                            child_execution_id: child_execution_id.clone(),
1537                            target_ffqn: FFQN_CHILD,
1538                            params,
1539                            result: Ok(()),
1540                        },
1541                    },
1542                },
1543            };
1544            let join_next = AppendRequest {
1545                created_at: current_time,
1546                event: ExecutionRequest::HistoryEvent {
1547                    event: HistoryEvent::JoinNext {
1548                        join_set_id: join_set_id.clone(),
1549                        run_expires_at: sim_clock.now(),
1550                        closing: false,
1551                        requested_ffqn: Some(FFQN_CHILD),
1552                    },
1553                },
1554            };
1555            db_pool
1556                .connection()
1557                .await
1558                .unwrap()
1559                .append_batch_create_new_execution(
1560                    current_time,
1561                    vec![join_set, child_exec_req, join_next],
1562                    parent_execution_id.clone(),
1563                    Version::new(2),
1564                    vec![child],
1565                    vec![],
1566                )
1567                .await
1568                .unwrap();
1569        }
1570
1571        let child_worker =
1572            Arc::new(SimpleWorker::with_single_result(worker_result).with_ffqn(FFQN_CHILD));
1573
1574        // execute the child
1575        tick_fn(
1576            ExecConfig {
1577                batch_size: 1,
1578                lock_expiry: LOCK_EXPIRY,
1579                tick_sleep: Duration::ZERO,
1580                component_id: ComponentId::dummy_activity(),
1581                task_limiter_global: None,
1582                task_limiter_local: None,
1583                executor_id: ExecutorId::generate(),
1584                retry_config: ComponentRetryConfig::ZERO,
1585                locking_strategy,
1586            },
1587            sim_clock.clone_box(),
1588            db_pool.clone(),
1589            child_worker,
1590            sim_clock.now(),
1591        )
1592        .await;
1593        if matches!(expected_child_err.kind, ExecutionFailureKind::TimedOut) {
1594            // In case of timeout, let the timers watcher handle it
1595            sim_clock.move_time_forward(LOCK_EXPIRY);
1596            expired_timers_watcher::tick(
1597                db_pool.connection().await.unwrap().as_ref(),
1598                sim_clock.now(),
1599            )
1600            .await
1601            .unwrap();
1602        }
1603        let child_log = db_pool
1604            .connection_test()
1605            .await
1606            .unwrap()
1607            .get(&ExecutionId::Derived(child_execution_id.clone()))
1608            .await
1609            .unwrap();
1610        assert!(child_log.pending_state.is_finished());
1611        assert_eq!(
1612            Version(2),
1613            child_log.next_version,
1614            "created = 0, locked = 1, with_single_result = 2"
1615        );
1616        assert_eq!(
1617            ExecutionRequest::Finished {
1618                retval: SupportedFunctionReturnValue::ExecutionFailure(expected_child_err),
1619                http_client_traces: None
1620            },
1621            child_log.last_event().event
1622        );
1623        let parent_log = db_pool
1624            .connection_test()
1625            .await
1626            .unwrap()
1627            .get(&parent_execution_id)
1628            .await
1629            .unwrap();
1630        assert_matches!(
1631            parent_log.pending_state,
1632            PendingState::PendingAt(PendingStatePendingAt {
1633                scheduled_at,
1634                last_lock: Some(LockedBy { executor_id: found_executor_id, run_id: _}),
1635            }) if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
1636            "parent should be back to pending"
1637        );
1638        let (found_join_set_id, found_child_execution_id, child_finished_version, found_result) = assert_matches!(
1639            parent_log.responses.last().map(|resp| &resp.event),
1640            Some(JoinSetResponseEventOuter{
1641                created_at: at,
1642                event: JoinSetResponseEvent{
1643                    join_set_id: found_join_set_id,
1644                    event: JoinSetResponse::ChildExecutionFinished {
1645                        child_execution_id: found_child_execution_id,
1646                        finished_version,
1647                        result: found_result,
1648                    }
1649                }
1650            })
1651             if *at == sim_clock.now()
1652            => (found_join_set_id, found_child_execution_id, finished_version, found_result)
1653        );
1654        assert_eq!(join_set_id, *found_join_set_id);
1655        assert_eq!(child_execution_id, *found_child_execution_id);
1656        assert_eq!(child_log.next_version, *child_finished_version);
1657        assert_matches!(
1658            found_result,
1659            SupportedFunctionReturnValue::ExecutionFailure(_)
1660        );
1661
1662        db_close.close().await;
1663    }
1664
1665    #[derive(Clone, Debug)]
1666    struct SleepyWorker {
1667        duration: Duration,
1668        result: SupportedFunctionReturnValue,
1669        exported: [FunctionMetadata; 1],
1670    }
1671
1672    #[async_trait]
1673    impl Worker for SleepyWorker {
1674        async fn run(&self, ctx: WorkerContext) -> WorkerResult {
1675            tokio::time::sleep(self.duration).await;
1676            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
1677                retval: self.result.clone(),
1678                version: ctx.version,
1679                http_client_traces: None,
1680            }))
1681        }
1682
1683        fn exported_functions_noext(&self) -> &[FunctionMetadata] {
1684            &self.exported
1685        }
1686    }
1687
1688    #[rstest]
1689    #[tokio::test]
1690    async fn hanging_lock_should_be_cleaned_and_execution_retried(
1691        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1692        locking_strategy: LockingStrategy,
1693    ) {
1694        set_up();
1695        let sim_clock = SimClock::default();
1696        let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1697        let lock_expiry = Duration::from_millis(100);
1698        let timeout_duration = Duration::from_millis(300);
1699        let retry_config = ComponentRetryConfig {
1700            max_retries: Some(1),
1701            retry_exp_backoff: timeout_duration,
1702        };
1703        let exec_config = ExecConfig {
1704            batch_size: 1,
1705            lock_expiry,
1706            tick_sleep: Duration::ZERO,
1707            component_id: ComponentId::dummy_activity(),
1708            task_limiter_global: None,
1709            task_limiter_local: None,
1710            executor_id: ExecutorId::generate(),
1711            retry_config,
1712            locking_strategy,
1713        };
1714
1715        let worker = Arc::new(SleepyWorker {
1716            duration: lock_expiry + Duration::from_millis(1), // sleep more than allowed by the lock expiry
1717            result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1718            exported: [FunctionMetadata {
1719                ffqn: FFQN_SOME,
1720                parameter_types: ParameterTypes::default(),
1721                return_type: RETURN_TYPE_DUMMY,
1722                extension: None,
1723                submittable: true,
1724            }],
1725        });
1726        // Create an execution
1727        let execution_id = ExecutionId::generate();
1728        let db_connection = db_pool.connection_test().await.unwrap();
1729        db_connection
1730            .create(CreateRequest {
1731                created_at: sim_clock.now(),
1732                execution_id: execution_id.clone(),
1733                ffqn: FFQN_SOME,
1734                params: Params::empty(),
1735                parent: None,
1736                metadata: concepts::ExecutionMetadata::empty(),
1737                scheduled_at: sim_clock.now(),
1738                component_id: ComponentId::dummy_activity(),
1739                deployment_id: DEPLOYMENT_ID_DUMMY,
1740                scheduled_by: None,
1741                paused: false,
1742            })
1743            .await
1744            .unwrap();
1745
1746        let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
1747        let executor = ExecTask::new_test(
1748            exec_config.clone(),
1749            worker,
1750            sim_clock.clone_box(),
1751            db_pool.clone(),
1752            ffqns,
1753        );
1754        let db_exec = db_pool.db_exec_conn().await.unwrap();
1755        let mut first_execution_progress = executor
1756            .tick(
1757                db_exec.as_ref(),
1758                sim_clock.now(),
1759                RunId::generate(),
1760                DEPLOYMENT_ID_DUMMY,
1761            )
1762            .await
1763            .unwrap();
1764        assert_eq!(1, first_execution_progress.executions.len());
1765        // Started hanging, wait for lock expiry.
1766        sim_clock.move_time_forward(lock_expiry);
1767        // cleanup should be called
1768        let now_after_first_lock_expiry = sim_clock.now();
1769        {
1770            debug!(now = %now_after_first_lock_expiry, "Expecting an expired lock");
1771            let cleanup_progress = executor
1772                .tick(
1773                    db_pool.db_exec_conn().await.unwrap().as_ref(),
1774                    now_after_first_lock_expiry,
1775                    RunId::generate(),
1776                    DEPLOYMENT_ID_DUMMY,
1777                )
1778                .await
1779                .unwrap();
1780            assert!(cleanup_progress.executions.is_empty());
1781        }
1782        {
1783            let expired_locks = expired_timers_watcher::tick(
1784                db_pool.connection().await.unwrap().as_ref(),
1785                now_after_first_lock_expiry,
1786            )
1787            .await
1788            .unwrap()
1789            .expired_locks;
1790            assert_eq!(1, expired_locks);
1791        }
1792        assert!(
1793            !first_execution_progress
1794                .executions
1795                .pop()
1796                .unwrap()
1797                .1
1798                .is_finished()
1799        );
1800
1801        let execution_log = db_connection.get(&execution_id).await.unwrap();
1802        let expected_first_timeout_expiry = now_after_first_lock_expiry + timeout_duration;
1803        assert_matches!(
1804            &execution_log.events.get(2).unwrap(),
1805            ExecutionEvent {
1806                event: ExecutionRequest::TemporarilyTimedOut { backoff_expires_at, .. },
1807                created_at: at,
1808                backtrace_id: None,
1809                version: Version(2),
1810            } if *at == now_after_first_lock_expiry && *backoff_expires_at == expected_first_timeout_expiry
1811        );
1812        assert_matches!(
1813            execution_log.pending_state,
1814            PendingState::PendingAt(PendingStatePendingAt {
1815                scheduled_at: found_scheduled_by,
1816                last_lock: Some(LockedBy {
1817                    executor_id: found_executor_id,
1818                    run_id: _,
1819                }),
1820            }) if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
1821        );
1822        sim_clock.move_time_forward(timeout_duration);
1823        let now_after_first_timeout = sim_clock.now();
1824        debug!(now = %now_after_first_timeout, "Second execution should hang again and result in a permanent timeout");
1825
1826        let mut second_execution_progress = executor
1827            .tick(
1828                db_pool.db_exec_conn().await.unwrap().as_ref(),
1829                now_after_first_timeout,
1830                RunId::generate(),
1831                DEPLOYMENT_ID_DUMMY,
1832            )
1833            .await
1834            .unwrap();
1835        assert_eq!(1, second_execution_progress.executions.len());
1836
1837        // Started hanging, wait for lock expiry.
1838        sim_clock.move_time_forward(lock_expiry);
1839        // cleanup should be called
1840        let now_after_second_lock_expiry = sim_clock.now();
1841        debug!(now = %now_after_second_lock_expiry, "Expecting the second lock to be expired");
1842        {
1843            let cleanup_progress = executor
1844                .tick(
1845                    db_pool.db_exec_conn().await.unwrap().as_ref(),
1846                    now_after_second_lock_expiry,
1847                    RunId::generate(),
1848                    DEPLOYMENT_ID_DUMMY,
1849                )
1850                .await
1851                .unwrap();
1852            assert!(cleanup_progress.executions.is_empty());
1853        }
1854        {
1855            let expired_locks = expired_timers_watcher::tick(
1856                db_pool.connection().await.unwrap().as_ref(),
1857                now_after_second_lock_expiry,
1858            )
1859            .await
1860            .unwrap()
1861            .expired_locks;
1862            assert_eq!(1, expired_locks);
1863        }
1864        assert!(
1865            !second_execution_progress
1866                .executions
1867                .pop()
1868                .unwrap()
1869                .1
1870                .is_finished()
1871        );
1872
1873        drop(db_connection);
1874        drop(executor);
1875        db_close.close().await;
1876    }
1877}
obeli_sk_executor/executor.rs

obeli_sk_executor/
executor.rs
