obeli_sk_executor/

executor.rs

use crate::worker::{
    ExecutionYieldReason, FatalError, RunFinished, Worker, WorkerContext, WorkerError,
    WorkerResult, WorkerResultOk,
};
use assert_matches::assert_matches;
use chrono::{DateTime, Utc};
use concepts::prefixed_ulid::{DeploymentId, RunId};
use concepts::storage::{
    AppendEventsToExecution, AppendRequest, AppendResponseToExecution, DbConnection,
    DbErrorGeneric, DbErrorWrite, DbErrorWriteNonRetriable, DbExecutor, DbPool, ExecutionLog,
    LockedExecution, Unlocked,
};
use concepts::time::{ClockFn, Sleep};
use concepts::{
    ComponentId, ComponentRetryConfig, ComponentType, FunctionMetadata, StrVariant,
    SupportedFunctionReturnValue,
};
use concepts::{ExecutionFailureKind, JoinSetId};
use concepts::{ExecutionId, FunctionFqn, prefixed_ulid::ExecutorId};
use concepts::{
    FinishedExecutionFailure,
    storage::{ExecutionRequest, Version},
};
use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};
use tokio::task::JoinHandle;
use tracing::{Instrument, Level, Span, debug, error, info, info_span, instrument, trace, warn};

#[derive(Debug, Clone)]
pub struct ExecConfig {
    pub lock_expiry: Duration,
    pub tick_sleep: Duration,
    pub batch_size: u32,
    pub component_id: ComponentId,
    pub task_limiter_global: Option<Arc<tokio::sync::Semaphore>>,
    pub task_limiter_local: Option<Arc<tokio::sync::Semaphore>>,
    pub executor_id: ExecutorId,
    pub retry_config: ComponentRetryConfig,
    pub locking_strategy: LockingStrategy,
}

pub struct ExecTask {
    worker: Arc<dyn Worker>,
    pub config: ExecConfig,
    clock_fn: Box<dyn ClockFn>, // Used for obtaining current time when the execution finishes.
    db_pool: Arc<dyn DbPool>,
    locking_strategy_holder: LockingStrategyHolder,
    worker_count_tx: tokio::sync::watch::Sender<usize>,
    execution_interrupt_watcher: tokio::sync::watch::Receiver<bool>,
}

#[derive(derive_more::Debug, Default)]
pub struct ExecutionProgress {
    #[debug(skip)]
    #[allow(dead_code)]
    executions: Vec<(ExecutionId, JoinHandle<()>)>,
}

impl ExecutionProgress {
    #[cfg(feature = "test")]
    pub async fn wait_for_tasks(self) -> Vec<ExecutionId> {
        let mut vec = Vec::new();
        for (exe, join_handle) in self.executions {
            vec.push(exe);
            join_handle.await.unwrap();
        }
        vec
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkerType {
    Activity,
    Workflow,
}

#[derive(derive_more::Debug)]
pub struct ExecutorTaskHandle {
    // Signals close() -> event loop to shut down
    #[debug(skip)]
    is_closing: Arc<AtomicBool>,
    #[debug(skip)]
    join_handle: JoinHandle<()>,
    component_id: ComponentId,
    executor_id: ExecutorId,
    deployment_id: DeploymentId,
    // Signals close() -> workers to shut down
    execution_interrupt_sender: tokio::sync::watch::Sender<bool>,
    /// Tracks the number of worker tasks currently in-flight.
    worker_count_rx: tokio::sync::watch::Receiver<usize>,
}

pub struct WorkerTasksHandle {
    component_id: ComponentId,
    executor_id: ExecutorId,
    deployment_id: DeploymentId,
    // Signals close() -> workers to shut down
    execution_interrupt_sender: tokio::sync::watch::Sender<bool>,
    /// Tracks the number of worker tasks currently in-flight.
    worker_count_rx: tokio::sync::watch::Receiver<usize>,
}

impl ExecutorTaskHandle {
    /// Shut down the executor task, return worker tasks handle untouched.
    #[instrument(name = "executor.close", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id,
        deployment_id = %self.deployment_id))]
    pub async fn close_outer_task(mut self) -> WorkerTasksHandle {
        debug!("Gracefully closing executor task");
        self.is_closing.store(true, Ordering::Relaxed);

        loop {
            tokio::select! {
                () = tokio::time::sleep(Duration::from_secs(5)) => {
                    info!("Waiting for executor to shut down");
                }
                _ = &mut self.join_handle => {
                    break;
                }
            }
        }
        let worker_tasks = *self.worker_count_rx.borrow();
        debug!("Gracefully closed executor task, {worker_tasks} workers are in progress");
        WorkerTasksHandle {
            component_id: self.component_id,
            executor_id: self.executor_id,
            deployment_id: self.deployment_id,
            execution_interrupt_sender: self.execution_interrupt_sender,
            worker_count_rx: self.worker_count_rx,
        }
    }

    #[must_use]
    pub fn component_id(&self) -> &ComponentId {
        &self.component_id
    }
}

impl WorkerTasksHandle {
    /// Signal to worker tasks, then wait for all tasks to finish.
    #[instrument(name = "WorkerTasksHandle.close", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id,
        deployment_id = %self.deployment_id))]
    pub async fn close(self) {
        debug!("Signaling worker tasks to unlock");
        let _ = self.execution_interrupt_sender.send(true);
        let mut worker_count_rx = self.worker_count_rx.clone();
        loop {
            tokio::select! {
                () = tokio::time::sleep(Duration::from_secs(1)) => {
                    info!("Waiting for {} workers to shut down", *self.worker_count_rx.borrow());
                }
                _ = worker_count_rx.wait_for(|&count| count == 0) => {
                    break;
                }
            }
        }
        debug!("All worker tasks are finished");
    }
}

#[cfg(feature = "test")]
pub fn extract_exported_ffqns_noext_test(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
    extract_exported_ffqns_noext(worker)
}

fn extract_exported_ffqns_noext(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
    worker
        .exported_functions_noext()
        .iter()
        .map(|FunctionMetadata { ffqn, .. }| ffqn.clone())
        .collect::<Arc<_>>()
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LockingStrategy {
    ByFfqns,
    ByComponentDigest,
    Auto,
}
impl LockingStrategy {
    fn holder(&self, ffqns: Arc<[FunctionFqn]>) -> LockingStrategyHolder {
        match self {
            LockingStrategy::ByFfqns => LockingStrategyHolder::ByFfqns(ffqns),
            LockingStrategy::ByComponentDigest => LockingStrategyHolder::ByComponentDigest,
            LockingStrategy::Auto => LockingStrategyHolder::Auto { ffqns },
        }
    }
}

enum LockingStrategyHolder {
    ByFfqns(Arc<[FunctionFqn]>),
    ByComponentDigest,
    Auto { ffqns: Arc<[FunctionFqn]> },
}

#[derive(Default)]
#[expect(dead_code)] // Stored permits limit semaphores until dropped.
struct TaskLimiterPermit {
    global: Option<tokio::sync::OwnedSemaphorePermit>,
    local: Option<tokio::sync::OwnedSemaphorePermit>,
}

impl ExecTask {
    #[cfg(feature = "test")]
    pub fn new_test(
        config: ExecConfig,
        worker: Arc<dyn Worker>,
        clock_fn: Box<dyn ClockFn>,
        db_pool: Arc<dyn DbPool>,
        ffqns: Arc<[FunctionFqn]>,
    ) -> (Self, tokio::sync::watch::Sender<bool>) {
        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
        let (close_tx, execution_interrupt_watcher) = tokio::sync::watch::channel(false);
        (
            ExecTask {
                worker,
                locking_strategy_holder: config.locking_strategy.holder(ffqns),
                config,
                clock_fn,
                db_pool,
                worker_count_tx,
                execution_interrupt_watcher,
            },
            close_tx,
        )
    }

    #[cfg(feature = "test")]
    pub fn new_all_ffqns_test(
        worker: Arc<dyn Worker>,
        config: ExecConfig,
        clock_fn: Box<dyn ClockFn>,
        db_pool: Arc<dyn DbPool>,
    ) -> (Self, tokio::sync::watch::Sender<bool>) {
        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
        let (close_tx, execution_interrupt_watcher) = tokio::sync::watch::channel(false);
        (
            Self {
                worker,
                locking_strategy_holder: config.locking_strategy.holder(ffqns),
                config,
                clock_fn,
                db_pool,
                worker_count_tx,
                execution_interrupt_watcher,
            },
            close_tx,
        )
    }

    #[cfg(feature = "test")]
    pub fn new_all_ffqns_test_with_interrupt_watcher(
        worker: Arc<dyn Worker>,
        config: ExecConfig,
        clock_fn: Box<dyn ClockFn>,
        db_pool: Arc<dyn DbPool>,
        execution_interrupt_watcher: tokio::sync::watch::Receiver<bool>,
    ) -> Self {
        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
        let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
        Self {
            worker,
            locking_strategy_holder: config.locking_strategy.holder(ffqns),
            config,
            clock_fn,
            db_pool,
            worker_count_tx,
            execution_interrupt_watcher,
        }
    }

    /// Spawn new tokio worker locking executions and starting worker tasks.
    pub fn spawn_new(
        deployment_id: DeploymentId,
        worker: Arc<dyn Worker>,
        config: ExecConfig,
        clock_fn: Box<dyn ClockFn>,
        db_pool: Arc<dyn DbPool>,
        sleep: impl Sleep + Clone + 'static,
    ) -> ExecutorTaskHandle {
        let is_closing = Arc::new(AtomicBool::default());
        let is_closing_inner = is_closing.clone();
        let ffqns = extract_exported_ffqns_noext(worker.as_ref());
        let component_id = config.component_id.clone();
        let executor_id = config.executor_id;
        let (worker_count_tx, worker_count_rx) = tokio::sync::watch::channel(0);
        let (execution_interrupt_sender, execution_interrupt_watcher) =
            tokio::sync::watch::channel(false);
        let join_handle = tokio::spawn(async move {
            debug!(executor_id = %config.executor_id, component_id = %config.component_id, "Spawned executor");
            let lock_strategy_holder = config.locking_strategy.holder(ffqns);
            let task = ExecTask {
                worker,
                config,
                db_pool,
                locking_strategy_holder: lock_strategy_holder,
                clock_fn: clock_fn.clone_box(),
                worker_count_tx,
                execution_interrupt_watcher,
            };
            let mut old_err = None;
            while !is_closing_inner.load(Ordering::Relaxed) {
                let res = task.db_pool.db_exec_conn().await;
                let res = log_err_if_new(res, &mut old_err);
                if let Ok(db_exec) = res {
                    let _ = task
                        .tick(
                            db_exec.as_ref(),
                            clock_fn.now(),
                            RunId::generate(),
                            deployment_id,
                        )
                        .await;
                    let timeout_fut = {
                        let sleep = sleep.clone();
                        Box::pin(async move { sleep.sleep(task.config.tick_sleep).await })
                    };
                    match &task.locking_strategy_holder {
                        LockingStrategyHolder::ByFfqns(ffqns) => {
                            db_exec
                                .wait_for_pending_by_ffqn(
                                    clock_fn.now(),
                                    ffqns.clone(),
                                    None,
                                    timeout_fut,
                                )
                                .await;
                        }
                        LockingStrategyHolder::ByComponentDigest => {
                            db_exec
                                .wait_for_pending_by_component_digest(
                                    clock_fn.now(),
                                    &task.config.component_id.component_digest,
                                    timeout_fut,
                                )
                                .await;
                        }
                        LockingStrategyHolder::Auto { ffqns } => {
                            db_exec
                                .wait_for_pending_by_ffqn(
                                    clock_fn.now(),
                                    ffqns.clone(),
                                    Some(task.config.component_id.component_digest.clone()),
                                    timeout_fut,
                                )
                                .await;
                        }
                    }
                } else {
                    sleep.sleep(task.config.tick_sleep).await;
                }
            }
        });
        ExecutorTaskHandle {
            is_closing,
            join_handle,
            component_id,
            executor_id,
            deployment_id,
            execution_interrupt_sender,
            worker_count_rx,
        }
    }

    fn acquire_task_permits(&self) -> Vec<TaskLimiterPermit> {
        let mut locks = Vec::with_capacity(
            usize::try_from(self.config.batch_size).expect("16 bit systems are unsupported"),
        );
        for _ in 0..self.config.batch_size {
            match (
                &self.config.task_limiter_global,
                &self.config.task_limiter_local,
            ) {
                (Some(global), Some(local)) => {
                    if let Ok(global) = global.clone().try_acquire_owned()
                        && let Ok(local) = local.clone().try_acquire_owned()
                    {
                        locks.push(TaskLimiterPermit {
                            global: Some(global),
                            local: Some(local),
                        });
                    } else {
                        break;
                    }
                }
                (Some(global), None) => {
                    if let Ok(global) = global.clone().try_acquire_owned() {
                        locks.push(TaskLimiterPermit {
                            global: Some(global),
                            local: None,
                        });
                    } else {
                        break;
                    }
                }
                (None, Some(local)) => {
                    if let Ok(local) = local.clone().try_acquire_owned() {
                        locks.push(TaskLimiterPermit {
                            global: None,
                            local: Some(local),
                        });
                    } else {
                        break;
                    }
                }
                (None, None) => {
                    locks.push(TaskLimiterPermit::default());
                }
            }
        }
        locks
    }

    #[cfg(feature = "test")]
    pub async fn tick_test(&self, executed_at: DateTime<Utc>, run_id: RunId) -> ExecutionProgress {
        use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;

        let db_exec = self.db_pool.db_exec_conn().await.unwrap();
        self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
            .await
            .unwrap()
    }

    #[cfg(feature = "test")]
    pub async fn tick_test_await(
        &self,
        executed_at: DateTime<Utc>,
        run_id: RunId,
    ) -> Vec<ExecutionId> {
        use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;

        let db_exec = self.db_pool.db_exec_conn().await.unwrap();
        self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
            .await
            .unwrap()
            .wait_for_tasks()
            .await
    }

    #[instrument(level = Level::TRACE, name = "executor.tick" skip_all, fields(executor_id = %self.config.executor_id, component_id = %self.config.component_id))]
    async fn tick(
        &self,
        db_exec: &dyn DbExecutor,
        executed_at: DateTime<Utc>,
        run_id: RunId,
        deployment_id: DeploymentId,
    ) -> Result<ExecutionProgress, DbErrorWrite> {
        let locked_executions = {
            let mut permits = self.acquire_task_permits();
            if permits.is_empty() {
                return Ok(ExecutionProgress::default());
            }
            let lock_expires_at = executed_at + self.config.lock_expiry;
            let batch_size = u32::try_from(permits.len()).expect("ExecConfig.batch_size is u32");
            let locked_executions = match &self.locking_strategy_holder {
                LockingStrategyHolder::Auto { ffqns } => {
                    db_exec
                        .lock_pending_by_ffqns_auto(
                            batch_size,
                            executed_at, // fetch expiring before now
                            ffqns.clone(),
                            executed_at, // created at
                            self.config.component_id.clone(),
                            deployment_id,
                            self.config.executor_id,
                            lock_expires_at,
                            run_id,
                            self.config.retry_config,
                        )
                        .await?
                }
                LockingStrategyHolder::ByFfqns(ffqns) => {
                    db_exec
                        .lock_pending_by_ffqns(
                            batch_size,
                            executed_at, // fetch expiring before now
                            ffqns.clone(),
                            executed_at, // created at
                            self.config.component_id.clone(),
                            deployment_id,
                            self.config.executor_id,
                            lock_expires_at,
                            run_id,
                            self.config.retry_config,
                        )
                        .await?
                }
                LockingStrategyHolder::ByComponentDigest => {
                    db_exec
                        .lock_pending_by_component_digest(
                            batch_size,
                            executed_at, // pending_at_or_sooner
                            &self.config.component_id,
                            deployment_id,
                            executed_at, // created at
                            self.config.executor_id,
                            lock_expires_at,
                            run_id,
                            self.config.retry_config,
                        )
                        .await?
                }
            };
            // Drop permits if too many were allocated.
            while permits.len() > locked_executions.len() {
                permits.pop();
            }
            assert_eq!(permits.len(), locked_executions.len());
            locked_executions.into_iter().zip(permits)
        };

        let mut executions = Vec::with_capacity(locked_executions.len());
        for (locked_execution, permit) in locked_executions {
            let execution_id = locked_execution.execution_id.clone();
            let join_handle = {
                let worker = self.worker.clone();
                let db_pool = self.db_pool.clone();
                let clock_fn = self.clock_fn.clone_box();
                let worker_span = info_span!(parent: None, "worker",
                    "otel.name" = format!("worker {}", locked_execution.ffqn),
                    %execution_id, %run_id,
                    ffqn = %locked_execution.ffqn,
                    executor_id = %self.config.executor_id,
                    component_id = %self.config.component_id,
                    %deployment_id,
                );
                locked_execution.metadata.enrich(&worker_span);
                let component_type = self.config.component_id.component_type;
                let worker_count_tx = self.worker_count_tx.clone();
                worker_count_tx.send_modify(|n| *n += 1);
                let execution_interrupt_watcher = self.execution_interrupt_watcher.clone();
                tokio::spawn({
                    let worker_span2 = worker_span.clone();
                    let retry_config = self.config.retry_config;
                    async move {
                        let _permit = permit;
                        let res = Self::run_worker(
                            component_type,
                            worker,
                            db_pool,
                            clock_fn,
                            locked_execution,
                            retry_config,
                            worker_span2,
                            execution_interrupt_watcher
                        )
                        .await;
                        debug!("run_worker finished with {res:?}");
                        if let Err(db_error) = res {
                            error!("Got db error `{db_error:?}`, expecting watcher to mark execution as timed out");
                        }
                        worker_count_tx.send_modify(|n| *n -= 1);
                    }
                    .instrument(worker_span)
                })
            };
            executions.push((execution_id, join_handle));
        }
        Ok(ExecutionProgress { executions })
    }

    #[expect(clippy::too_many_arguments)]
    async fn run_worker(
        component_type: ComponentType,
        worker: Arc<dyn Worker>,
        db_pool: Arc<dyn DbPool>,
        clock_fn: Box<dyn ClockFn>,
        locked_execution: LockedExecution,
        retry_config: ComponentRetryConfig,
        worker_span: Span,
        execution_interrupt_watcher: tokio::sync::watch::Receiver<bool>,
    ) -> Result<(), DbErrorWrite> {
        debug!("Worker::run starting");
        trace!(
            version = %locked_execution.next_version,
            params = ?locked_execution.params,
            event_history = ?locked_execution.event_history,
            "Worker::run starting"
        );
        let can_be_retried = ExecutionLog::can_be_retried_after(
            locked_execution.intermittent_event_count + 1,
            retry_config.max_retries,
            retry_config.retry_exp_backoff,
        );
        let unlock_expiry_on_limit_reached =
            ExecutionLog::compute_retry_duration_when_retrying_forever(
                locked_execution.intermittent_event_count + 1,
                retry_config.retry_exp_backoff,
            );
        let parent = locked_execution.parent.clone();
        let execution_id = locked_execution.execution_id.clone();
        let ctx = WorkerContext {
            execution_id: locked_execution.execution_id.clone(),
            metadata: locked_execution.metadata,
            component_digest: locked_execution.component_digest,
            ffqn: locked_execution.ffqn,
            params: locked_execution.params,
            event_history: locked_execution.event_history,
            responses: locked_execution.responses,
            parent: parent.clone(),
            version: locked_execution.next_version,
            can_be_retried: can_be_retried.is_some(),
            locked_event: locked_execution.locked_event,
            worker_span,
            execution_interrupt_watcher,
        };
        let worker_result = worker.run(ctx).await;
        debug!("Worker::run finished {worker_result:?}");
        let result_obtained_at = clock_fn.now();
        if component_type == ComponentType::Activity
            && matches!(
                &worker_result,
                WorkerResult::Err(WorkerError::FatalError(FatalError::Cancelled, _))
            )
        {
            let db_conn = db_pool.connection().await?;
            if let Some(append) = Self::cancelled_activity_to_append(
                db_conn.as_ref(),
                execution_id,
                result_obtained_at,
            )
            .await?
            {
                trace!("Appending activity cancellation {append:?}");
                append.append(db_conn.as_ref()).await?;
            }
            return Ok(());
        }
        match Self::worker_result_to_execution_event(
            component_type,
            execution_id,
            worker_result,
            result_obtained_at,
            parent,
            can_be_retried,
            unlock_expiry_on_limit_reached,
        )? {
            Some(append) => {
                debug!("Appending {append:?}");
                let db_exec = db_pool.db_exec_conn().await?;
                append.append(db_exec.as_ref()).await
            }
            None => Ok(()),
        }
    }

    async fn cancelled_activity_to_append(
        db_conn: &dyn DbConnection,
        execution_id: ExecutionId,
        result_obtained_at: DateTime<Utc>,
    ) -> Result<Option<Append>, DbErrorWrite> {
        let log = db_conn
            .get(&execution_id)
            .await
            .map_err(DbErrorWrite::from)?;
        if log.is_finished() {
            return Ok(None);
        }
        if !log.pending_state.is_cancelling() {
            warn!(
                %execution_id,
                "Activity reported cancellation after teardown, but storage is not cancelling"
            );
            return Ok(None);
        }
        let result = SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure {
            reason: None,
            kind: ExecutionFailureKind::Cancelled,
            detail: None,
        });
        let child_finished =
            log.parent().map(
                |(parent_execution_id, parent_join_set)| ChildFinishedResponse {
                    parent_execution_id,
                    parent_join_set,
                    result: result.clone(),
                },
            );
        Ok(Some(Append {
            created_at: result_obtained_at,
            primary_event: AppendRequest {
                created_at: result_obtained_at,
                event: ExecutionRequest::Finished {
                    retval: result,
                    http_client_traces: None,
                },
            },
            execution_id,
            version: log.next_version,
            child_finished,
        }))
    }

    /// Map the `WorkerError` to an optional append event
    fn worker_result_to_execution_event(
        component_type: ComponentType,
        execution_id: ExecutionId,
        worker_result: WorkerResult,
        result_obtained_at: DateTime<Utc>,
        parent: Option<(ExecutionId, JoinSetId)>,
        can_be_retried: Option<Duration>,
        unlock_expiry_on_limit_reached: Duration,
    ) -> Result<Option<Append>, DbErrorWrite> {
        Ok(match worker_result {
            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                retval: ref retval @ SupportedFunctionReturnValue::Err(ref result_err),
                version,
                http_client_traces,
            })) if component_type == ComponentType::Activity
                && can_be_retried.is_some()
                && !retval.is_permanent_variant() =>
            {
                // Interpret returned `err` variant as a retry request, unless it is a permanent variant
                let detail = serde_json::to_string(result_err)
                    .expect("SupportedFunctionReturnValue should be serializable to JSON");
                let duration = can_be_retried.expect(
                    "ActivityReturnedError must not be returned when retries are exhausted",
                );
                let expires_at = result_obtained_at + duration;
                debug!("Retrying ActivityReturnedError after {duration:?} at {expires_at}");
                let primary_event = ExecutionRequest::TemporarilyFailed {
                    backoff_expires_at: expires_at,
                    reason: StrVariant::Static("activity finished with error"),
                    detail: Some(detail),
                    http_client_traces,
                };
                Some(Append {
                    created_at: result_obtained_at,
                    primary_event: AppendRequest {
                        created_at: result_obtained_at,
                        event: primary_event,
                    },
                    execution_id,
                    version,
                    child_finished: None,
                })
            }

            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                retval: result,
                version,
                http_client_traces,
            })) => {
                info!("Execution finished: {result}");
                let child_finished =
                    parent.map(
                        |(parent_execution_id, parent_join_set)| ChildFinishedResponse {
                            parent_execution_id,
                            parent_join_set,
                            result: result.clone(),
                        },
                    );
                let primary_event = AppendRequest {
                    created_at: result_obtained_at,
                    event: ExecutionRequest::Finished {
                        retval: result,
                        http_client_traces,
                    },
                };

                Some(Append {
                    created_at: result_obtained_at,
                    primary_event,
                    execution_id,
                    version,
                    child_finished,
                })
            }

            WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher) => None,

            WorkerResult::Err(err) => {
                let reason_generic = err.to_string(); // Override with err's reason if no information is lost.

                let (primary_event, child_finished, version) = match err {
                    WorkerError::ExecutionYielded { version, reason } => {
                        let primary_event = ExecutionRequest::Unlocked(Unlocked {
                            unlocked_at: result_obtained_at,
                            reason: match reason {
                                ExecutionYieldReason::ExecutorClosing => "executor closing".into(),
                                ExecutionYieldReason::WorkflowEventLimitReached => {
                                    "workflow event limit reached".into()
                                }
                            },
                        });
                        (primary_event, None, version)
                    }
                    WorkerError::TemporaryTimeout {
                        http_client_traces,
                        version,
                    } => {
                        if let Some(duration) = can_be_retried {
                            let backoff_expires_at = result_obtained_at + duration;
                            info!(
                                "Temporary timeout, retrying after {duration:?} at {backoff_expires_at}"
                            );
                            (
                                ExecutionRequest::TemporarilyTimedOut {
                                    backoff_expires_at,
                                    http_client_traces,
                                },
                                None,
                                version,
                            )
                        } else {
                            info!("Execution timed out");
                            let result = SupportedFunctionReturnValue::ExecutionFailure(
                                FinishedExecutionFailure {
                                    kind: ExecutionFailureKind::TimedOut,
                                    reason: None,
                                    detail: None,
                                },
                            );
                            let child_finished =
                                parent.map(|(parent_execution_id, parent_join_set)| {
                                    ChildFinishedResponse {
                                        parent_execution_id,
                                        parent_join_set,
                                        result: result.clone(),
                                    }
                                });
                            (
                                ExecutionRequest::Finished {
                                    retval: result,
                                    http_client_traces,
                                },
                                child_finished,
                                version,
                            )
                        }
                    }
                    WorkerError::DbError(db_error) => {
                        return Err(db_error);
                    }
                    WorkerError::ActivityTrap {
                        reason: _, // reason_generic contains trap_kind + reason
                        trap_kind,
                        detail,
                        version,
                        http_client_traces,
                    } => {
                        if let Some(duration) = can_be_retried {
                            let expires_at = result_obtained_at + duration;
                            debug!(
                                "Retrying activity with `{trap_kind}` execution after {duration:?} at {expires_at}"
                            );
                            (
                                ExecutionRequest::TemporarilyFailed {
                                    reason: StrVariant::from(reason_generic),
                                    backoff_expires_at: expires_at,
                                    detail,
                                    http_client_traces,
                                },
                                None,
                                version,
                            )
                        } else {
                            info!(
                                "Activity with `{trap_kind}` marked as permanent failure - {reason_generic}"
                            );
                            let result = SupportedFunctionReturnValue::ExecutionFailure(
                                FinishedExecutionFailure {
                                    reason: Some(reason_generic),
                                    kind: ExecutionFailureKind::Uncategorized,
                                    detail,
                                },
                            );
                            let child_finished =
                                parent.map(|(parent_execution_id, parent_join_set)| {
                                    ChildFinishedResponse {
                                        parent_execution_id,
                                        parent_join_set,
                                        result: result.clone(),
                                    }
                                });
                            (
                                ExecutionRequest::Finished {
                                    retval: result,
                                    http_client_traces,
                                },
                                child_finished,
                                version,
                            )
                        }
                    }
                    WorkerError::LimitReached {
                        reason,
                        version: new_version,
                    } => {
                        let expires_at = result_obtained_at + unlock_expiry_on_limit_reached;
                        warn!(
                            "Limit reached: {reason}, unlocking after {unlock_expiry_on_limit_reached:?} at {expires_at}"
                        );
                        (
                            ExecutionRequest::Unlocked(Unlocked {
                                unlocked_at: expires_at,
                                reason: StrVariant::from(reason),
                            }),
                            None,
                            new_version,
                        )
                    }
                    WorkerError::FatalError(FatalError::Cancelled, _version) => {
                        unreachable!(
                            "activity Cancelled is handled before this match; workflows are cancelled via the driver and never return FatalError::Cancelled"
                        )
                    }
                    WorkerError::FatalError(fatal_error, version) => {
                        warn!("Fatal worker error - {fatal_error:?}");
                        let result = SupportedFunctionReturnValue::ExecutionFailure(
                            FinishedExecutionFailure::from(fatal_error),
                        );
                        let child_finished =
                            parent.map(|(parent_execution_id, parent_join_set)| {
                                ChildFinishedResponse {
                                    parent_execution_id,
                                    parent_join_set,
                                    result: result.clone(),
                                }
                            });
                        (
                            ExecutionRequest::Finished {
                                retval: result,
                                http_client_traces: None,
                            },
                            child_finished,
                            version,
                        )
                    }
                };
                Some(Append {
                    created_at: result_obtained_at,
                    primary_event: AppendRequest {
                        created_at: result_obtained_at,
                        event: primary_event,
                    },
                    execution_id,
                    version,
                    child_finished,
                })
            }
        })
    }
}

#[derive(Debug, Clone)]
pub(crate) struct ChildFinishedResponse {
    pub(crate) parent_execution_id: ExecutionId,
    pub(crate) parent_join_set: JoinSetId,
    pub(crate) result: SupportedFunctionReturnValue,
}

#[derive(Debug, Clone)]
pub(crate) struct Append {
    pub(crate) created_at: DateTime<Utc>,
    pub(crate) primary_event: AppendRequest,
    pub(crate) execution_id: ExecutionId,
    pub(crate) version: Version,
    pub(crate) child_finished: Option<ChildFinishedResponse>,
}

impl Append {
    pub(crate) async fn append(self, db_exec: &dyn DbExecutor) -> Result<(), DbErrorWrite> {
        if let Some(child_finished) = self.child_finished {
            assert_matches!(
                &self.primary_event,
                AppendRequest {
                    event: ExecutionRequest::Finished { .. },
                    ..
                }
            );
            let child_execution_id = assert_matches!(self.execution_id.clone(), ExecutionId::Derived(derived) => derived);
            let events = AppendEventsToExecution {
                execution_id: self.execution_id,
                version: self.version.clone(),
                batch: vec![self.primary_event],
            };
            let response = AppendResponseToExecution {
                parent_execution_id: child_finished.parent_execution_id,
                created_at: self.created_at,
                join_set_id: child_finished.parent_join_set,
                child_execution_id,
                finished_version: self.version, // Since self.primary_event is a finished event, the version will remain the same.
                result: child_finished.result,
            };

            db_exec
                .append_batch_respond_to_parent(events, response, self.created_at)
                .await
                .map(|_| ())
        } else {
            let res = db_exec
                .append(self.execution_id, self.version, self.primary_event)
                .await;
            match res {
                Ok(_) => Ok(()),
                Err(DbErrorWrite::NonRetriable(
                    DbErrorWriteNonRetriable::UnlockedCannotBeAppended(reason),
                )) => {
                    // Ignore unsuccessful `Unlocked`
                    debug!("Ignoring unlock failed - {reason}");
                    Ok(())
                }
                Err(err) => Err(err),
            }
        }
    }
}

fn log_err_if_new<T>(
    res: Result<T, DbErrorGeneric>,
    old_err: &mut Option<DbErrorGeneric>,
) -> Result<T, ()> {
    match (res, &old_err) {
        (Ok(ok), _) => {
            *old_err = None;
            Ok(ok)
        }
        (Err(err), Some(old)) if err == *old => Err(()),
        (Err(err), _) => {
            warn!("Tick failed: {err:?}");
            *old_err = Some(err);
            Err(())
        }
    }
}

#[cfg(any(test, feature = "test"))]
pub mod simple_worker {
    use crate::worker::{Worker, WorkerContext, WorkerResult};
    use async_trait::async_trait;
    use concepts::{
        FunctionFqn, FunctionMetadata, ParameterTypes, RETURN_TYPE_DUMMY,
        storage::{HistoryEvent, Version},
    };
    use indexmap::IndexMap;
    use std::sync::Arc;
    use tracing::trace;

    pub(crate) const FFQN_SOME: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn");
    pub type SimpleWorkerResultMap =
        Arc<std::sync::Mutex<IndexMap<Version, (Vec<HistoryEvent>, WorkerResult)>>>;

    #[derive(Clone, Debug)]
    pub struct SimpleWorker {
        pub worker_results_rev: SimpleWorkerResultMap,
        pub ffqn: FunctionFqn,
        exported: [FunctionMetadata; 1],
    }

    impl SimpleWorker {
        #[must_use]
        pub fn with_single_result(res: WorkerResult) -> Self {
            Self::with_worker_results_rev(Arc::new(std::sync::Mutex::new(IndexMap::from([(
                Version::new(2),
                (vec![], res),
            )]))))
        }

        #[must_use]
        pub fn with_ffqn(self, ffqn: FunctionFqn) -> Self {
            Self {
                worker_results_rev: self.worker_results_rev,
                exported: [FunctionMetadata {
                    ffqn: ffqn.clone(),
                    parameter_types: ParameterTypes::default(),
                    return_type: RETURN_TYPE_DUMMY,
                    extension: None,
                    submittable: true,
                }],
                ffqn,
            }
        }

        #[must_use]
        pub fn with_worker_results_rev(worker_results_rev: SimpleWorkerResultMap) -> Self {
            Self {
                worker_results_rev,
                ffqn: FFQN_SOME,
                exported: [FunctionMetadata {
                    ffqn: FFQN_SOME,
                    parameter_types: ParameterTypes::default(),
                    return_type: RETURN_TYPE_DUMMY,
                    extension: None,
                    submittable: true,
                }],
            }
        }
    }

    #[async_trait]
    impl Worker for SimpleWorker {
        async fn run(&self, ctx: WorkerContext) -> WorkerResult {
            let (expected_version, (expected_eh, worker_result)) =
                self.worker_results_rev.lock().unwrap().pop().unwrap();
            trace!(%expected_version, version = %ctx.version, ?expected_eh, eh = ?ctx.event_history, "Running SimpleWorker");
            assert_eq!(expected_version, ctx.version);
            assert_eq!(
                expected_eh,
                ctx.event_history
                    .iter()
                    .map(|(event, _version)| event.clone())
                    .collect::<Vec<_>>()
            );
            worker_result
        }

        fn exported_functions_noext(&self) -> &[FunctionMetadata] {
            &self.exported
        }
    }
}

#[cfg(test)]
mod tests {
    use self::simple_worker::SimpleWorker;
    use super::*;
    use crate::{expired_timers_watcher, worker::WorkerResult};
    use assert_matches::assert_matches;
    use async_trait::async_trait;
    use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
    use concepts::storage::{
        CreateRequest, DbConnectionTest, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent,
    };
    use concepts::storage::{DbPoolCloseable, LockedBy};
    use concepts::storage::{
        ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState, PendingStatePendingAt,
    };
    use concepts::time::ConstClock;
    use concepts::{
        FunctionMetadata, JoinSetKind, ParameterTypes, Params, RETURN_TYPE_DUMMY,
        SUPPORTED_RETURN_VALUE_OK_EMPTY, StrVariant, SupportedFunctionReturnValue, TrapKind,
    };
    use db_tests::Database;
    use indexmap::IndexMap;
    use rstest::rstest;
    use simple_worker::FFQN_SOME;
    use std::{fmt::Debug, future::Future, ops::Deref, sync::Arc};
    use test_db_macro::expand_enum_database;
    use test_utils::set_up;
    use test_utils::sim_clock::SimClock;

    pub(crate) const FFQN_CHILD: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn-child");

    async fn tick_fn<W: Worker + Debug>(
        config: ExecConfig,
        clock_fn: Box<dyn ClockFn>,
        db_pool: Arc<dyn DbPool>,
        worker: Arc<W>,
        executed_at: DateTime<Utc>,
    ) -> Vec<ExecutionId> {
        trace!("Ticking with {worker:?}");
        let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
        let (executor, _close_tx) = ExecTask::new_test(config, worker, clock_fn, db_pool, ffqns);
        executor
            .tick_test_await(executed_at, RunId::generate())
            .await
    }

    #[expand_enum_database]
    #[rstest]
    #[tokio::test]
    async fn execute_simple_lifecycle_tick_based(
        database: Database,
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        set_up();
        let created_at = DateTime::UNIX_EPOCH;
        let (_guard, db_pool, db_close) = database.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();
        execute_simple_lifecycle_tick_based_inner(
            db_connection.as_ref(),
            db_pool.clone(),
            Box::new(ConstClock(created_at)),
            locking_strategy,
        )
        .await;
        drop(db_connection);
        db_close.close().await;
    }

    async fn execute_simple_lifecycle_tick_based_inner(
        db_connection: &dyn DbConnectionTest,
        db_pool: Arc<dyn DbPool>,
        clock_fn: Box<dyn ClockFn>,
        locking_strategy: LockingStrategy,
    ) {
        let created_at = clock_fn.now();
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::from_millis(100),
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };

        let execution_log = create_and_tick(
            CreateAndTickConfig {
                execution_id: ExecutionId::generate(),
                created_at,
                executed_at: created_at,
            },
            clock_fn,
            db_connection,
            db_pool,
            exec_config,
            Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
                WorkerResultOk::RunFinished(RunFinished {
                    retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    version: Version::new(2),
                    http_client_traces: None,
                }),
            ))),
            tick_fn,
        )
        .await;
        assert_matches!(
            execution_log.events.get(2).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Finished {
                    retval: SupportedFunctionReturnValue::Ok(None),
                    http_client_traces: None
                },
                created_at: _,
                backtrace_id: None,
                version: Version(2),
            }
        );
    }

    #[rstest]
    #[tokio::test]
    async fn execute_simple_lifecycle_task_based_mem(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        set_up();
        let created_at = DateTime::UNIX_EPOCH;
        let clock_fn = Box::new(ConstClock(created_at));
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };

        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
            WorkerResultOk::RunFinished(RunFinished {
                retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                version: Version::new(2),
                http_client_traces: None,
            }),
        )));
        let db_connection = db_pool.connection_test().await.unwrap();

        let execution_log = create_and_tick(
            CreateAndTickConfig {
                execution_id: ExecutionId::generate(),
                created_at,
                executed_at: created_at,
            },
            clock_fn,
            db_connection.as_ref(),
            db_pool,
            exec_config,
            worker,
            tick_fn,
        )
        .await;
        assert_matches!(
            execution_log.events.get(2).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Finished {
                    retval: SupportedFunctionReturnValue::Ok(None),
                    http_client_traces: None
                },
                created_at: _,
                backtrace_id: None,
                version: Version(2),
            }
        );
        db_close.close().await;
    }

    struct CreateAndTickConfig {
        execution_id: ExecutionId,
        created_at: DateTime<Utc>,
        executed_at: DateTime<Utc>,
    }

    async fn create_and_tick<
        W: Worker,
        T: FnMut(ExecConfig, Box<dyn ClockFn>, Arc<dyn DbPool>, Arc<W>, DateTime<Utc>) -> F,
        F: Future<Output = Vec<ExecutionId>>,
    >(
        config: CreateAndTickConfig,
        clock_fn: Box<dyn ClockFn>,
        db_connection: &dyn DbConnectionTest,
        db_pool: Arc<dyn DbPool>,
        exec_config: ExecConfig,
        worker: Arc<W>,
        mut tick: T,
    ) -> ExecutionLog {
        // Create an execution
        db_connection
            .create(CreateRequest {
                created_at: config.created_at,
                execution_id: config.execution_id.clone(),
                ffqn: FFQN_SOME,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: config.created_at,
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        // execute!
        tick(exec_config, clock_fn, db_pool, worker, config.executed_at).await;
        let execution_log = db_connection.get(&config.execution_id).await.unwrap();
        debug!("Execution history after tick: {execution_log:?}");
        // check that DB contains Created and Locked events.
        let actually_created_at = assert_matches!(
            execution_log.events.first().unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Created { .. },
                created_at: actually_created_at,
                backtrace_id: None,
                version: Version(0),
            }
            => *actually_created_at
        );
        assert_eq!(config.created_at, actually_created_at);
        let locked_at = assert_matches!(
            execution_log.events.get(1).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Locked { .. },
                created_at: locked_at,
                backtrace_id: None,
                version: Version(1),
            } if config.created_at <= *locked_at
            => *locked_at
        );
        assert_matches!(execution_log.events.get(2).unwrap(), ExecutionEvent {
            event: _,
            created_at: executed_at,
            backtrace_id: None,
            version: Version(2),
        } if *executed_at >= locked_at);
        execution_log
    }

    #[rstest]
    #[tokio::test]
    async fn activity_trap_should_trigger_an_execution_retry(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let retry_exp_backoff = Duration::from_millis(100);
        let retry_config = ComponentRetryConfig {
            max_retries: Some(1),
            retry_exp_backoff,
        };
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config,
            locking_strategy,
        };
        let expected_reason = "error reason";
        let expected_detail = "error detail";
        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
            WorkerError::ActivityTrap {
                reason: expected_reason.to_string(),
                trap_kind: concepts::TrapKind::Trap,
                detail: Some(expected_detail.to_string()),
                version: Version::new(2),
                http_client_traces: None,
            },
        )));
        debug!(now = %sim_clock.now(), "Creating an execution that should fail");
        let db_connection = db_pool.connection_test().await.unwrap();
        let execution_log = create_and_tick(
            CreateAndTickConfig {
                execution_id: ExecutionId::generate(),
                created_at: sim_clock.now(),
                executed_at: sim_clock.now(),
            },
            sim_clock.clone_box(),
            db_connection.as_ref(),
            db_pool.clone(),
            exec_config.clone(),
            worker,
            tick_fn,
        )
        .await;
        assert_eq!(3, execution_log.events.len());
        {
            let (reason, detail, at, expires_at) = assert_matches!(
                &execution_log.events.get(2).unwrap(),
                ExecutionEvent {
                    event: ExecutionRequest::TemporarilyFailed {
                        reason,
                        detail,
                        backoff_expires_at,
                        http_client_traces: None,
                    },
                    created_at: at,
                    backtrace_id: None,
                    version: Version(2),
                }
                => (reason, detail, *at, *backoff_expires_at)
            );
            assert_eq!(format!("activity trap: {expected_reason}"), reason.deref());
            assert_eq!(Some(expected_detail), detail.as_deref());
            assert_eq!(at, sim_clock.now());
            assert_eq!(sim_clock.now() + retry_config.retry_exp_backoff, expires_at);
        }
        let worker = Arc::new(SimpleWorker::with_worker_results_rev(Arc::new(
            std::sync::Mutex::new(IndexMap::from([(
                Version::new(4),
                (
                    vec![],
                    WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                        retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                        version: Version::new(4),
                        http_client_traces: None,
                    })),
                ),
            )])),
        )));
        // noop until `retry_exp_backoff` expires
        assert!(
            tick_fn(
                exec_config.clone(),
                sim_clock.clone_box(),
                db_pool.clone(),
                worker.clone(),
                sim_clock.now(),
            )
            .await
            .is_empty()
        );
        // tick again to finish the execution
        sim_clock.move_time_forward(retry_config.retry_exp_backoff);
        tick_fn(
            exec_config,
            sim_clock.clone_box(),
            db_pool.clone(),
            worker,
            sim_clock.now(),
        )
        .await;
        let execution_log = {
            let db_connection = db_pool.connection_test().await.unwrap();
            db_connection
                .get(&execution_log.execution_id)
                .await
                .unwrap()
        };
        debug!(now = %sim_clock.now(), "Execution history after second tick: {execution_log:?}");
        assert_matches!(
            execution_log.events.get(3).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Locked { .. },
                created_at: at,
                backtrace_id: None,
                version: Version(3),
            } if *at == sim_clock.now()
        );
        assert_matches!(
            execution_log.events.get(4).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Finished {
                    retval: SupportedFunctionReturnValue::Ok(None),
                    http_client_traces: None
                },
                created_at: finished_at,
                backtrace_id: None,
                version: Version(4),
            } if *finished_at == sim_clock.now()
        );
        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn activity_trap_should_not_be_retried_if_no_retries_are_set(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        set_up();
        let created_at = DateTime::UNIX_EPOCH;
        let clock_fn = Box::new(ConstClock(created_at));
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };

        let reason = "error reason";
        let expected_reason = format!("activity trap: {reason}");
        let expected_detail = "error detail";
        let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
            WorkerError::ActivityTrap {
                reason: reason.to_string(),
                trap_kind: concepts::TrapKind::Trap,
                detail: Some(expected_detail.to_string()),
                version: Version::new(2),
                http_client_traces: None,
            },
        )));
        let execution_log = create_and_tick(
            CreateAndTickConfig {
                execution_id: ExecutionId::generate(),
                created_at,
                executed_at: created_at,
            },
            clock_fn,
            db_pool.connection_test().await.unwrap().as_ref(),
            db_pool.clone(),
            exec_config.clone(),
            worker,
            tick_fn,
        )
        .await;
        assert_eq!(3, execution_log.events.len());
        let (reason, kind, detail) = assert_matches!(
            &execution_log.events.get(2).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::Finished{
                    retval: SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure{reason, kind, detail}),
                    http_client_traces: None
                },
                created_at: at,
                backtrace_id: None,
                version: Version(2),
            } if *at == created_at
            => (reason, kind, detail)
        );

        assert_eq!(Some(expected_reason), *reason);
        assert_eq!(Some(expected_detail), detail.as_deref());
        assert_eq!(ExecutionFailureKind::Uncategorized, *kind);

        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn child_execution_permanently_failed_should_notify_parent_permanent_failure(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        let worker_error = WorkerError::ActivityTrap {
            reason: "error reason".to_string(),
            trap_kind: TrapKind::Trap,
            detail: Some("detail".to_string()),
            version: Version::new(2),
            http_client_traces: None,
        };
        let expected_child_err = FinishedExecutionFailure {
            kind: ExecutionFailureKind::Uncategorized,
            reason: Some("activity trap: error reason".to_string()),
            detail: Some("detail".to_string()),
        };
        child_execution_permanently_failed_should_notify_parent(
            WorkerResult::Err(worker_error),
            expected_child_err,
            locking_strategy,
        )
        .await;
    }

    #[rstest]
    #[tokio::test]
    async fn child_execution_permanently_failed_handled_by_watcher_should_notify_parent_timeout(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        let expected_child_err = FinishedExecutionFailure {
            kind: ExecutionFailureKind::TimedOut,
            reason: None,
            detail: None,
        };
        child_execution_permanently_failed_should_notify_parent(
            WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher),
            expected_child_err,
            locking_strategy,
        )
        .await;
    }

    async fn child_execution_permanently_failed_should_notify_parent(
        worker_result: WorkerResult,
        expected_child_err: FinishedExecutionFailure,
        locking_strategy: LockingStrategy,
    ) {
        use concepts::storage::JoinSetResponseEventOuter;
        const LOCK_EXPIRY: Duration = Duration::from_secs(1);

        set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;

        let parent_worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
            WorkerResultOk::DbUpdatedByWorkerOrWatcher,
        )));
        let parent_execution_id = ExecutionId::generate();
        db_pool
            .connection()
            .await
            .unwrap()
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: parent_execution_id.clone(),
                ffqn: FFQN_SOME,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let parent_executor_id = ExecutorId::generate();
        tick_fn(
            ExecConfig {
                batch_size: 1,
                lock_expiry: LOCK_EXPIRY,
                tick_sleep: Duration::ZERO,
                component_id: ComponentId::dummy_activity(),
                task_limiter_global: None,
                task_limiter_local: None,
                executor_id: parent_executor_id,
                retry_config: ComponentRetryConfig::ZERO,
                locking_strategy,
            },
            sim_clock.clone_box(),
            db_pool.clone(),
            parent_worker,
            sim_clock.now(),
        )
        .await;

        let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let child_execution_id = parent_execution_id.next_level(&join_set_id);
        // executor does not append anything, this should have been written by the worker:
        {
            let params = Params::empty();
            let child = CreateRequest {
                created_at: sim_clock.now(),
                execution_id: ExecutionId::Derived(child_execution_id.clone()),
                ffqn: FFQN_CHILD,
                params: params.clone(),
                parent: Some((parent_execution_id.clone(), join_set_id.clone())),
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            };
            let current_time = sim_clock.now();
            let join_set = AppendRequest {
                created_at: current_time,
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetCreate {
                        join_set_id: join_set_id.clone(),
                    },
                },
            };
            let child_exec_req = AppendRequest {
                created_at: current_time,
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::ChildExecutionRequest {
                            child_execution_id: child_execution_id.clone(),
                            target_ffqn: FFQN_CHILD,
                            params,
                            result: Ok(()),
                        },
                    },
                },
            };
            let join_next = AppendRequest {
                created_at: current_time,
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinNext {
                        join_set_id: join_set_id.clone(),
                        run_expires_at: sim_clock.now(),
                        closing: false,
                        requested_ffqn: Some(FFQN_CHILD),
                    },
                },
            };
            db_pool
                .connection()
                .await
                .unwrap()
                .append_batch_create_new_execution(
                    current_time,
                    vec![join_set, child_exec_req, join_next],
                    parent_execution_id.clone(),
                    Version::new(2),
                    vec![child],
                    vec![],
                )
                .await
                .unwrap();
        }

        let child_worker =
            Arc::new(SimpleWorker::with_single_result(worker_result).with_ffqn(FFQN_CHILD));

        // execute the child
        tick_fn(
            ExecConfig {
                batch_size: 1,
                lock_expiry: LOCK_EXPIRY,
                tick_sleep: Duration::ZERO,
                component_id: ComponentId::dummy_activity(),
                task_limiter_global: None,
                task_limiter_local: None,
                executor_id: ExecutorId::generate(),
                retry_config: ComponentRetryConfig::ZERO,
                locking_strategy,
            },
            sim_clock.clone_box(),
            db_pool.clone(),
            child_worker,
            sim_clock.now(),
        )
        .await;
        if matches!(expected_child_err.kind, ExecutionFailureKind::TimedOut) {
            // In case of timeout, let the timers watcher handle it
            sim_clock.move_time_forward(LOCK_EXPIRY);
            expired_timers_watcher::tick(
                db_pool.connection().await.unwrap().as_ref(),
                sim_clock.now(),
            )
            .await
            .unwrap();
        }
        let child_log = db_pool
            .connection_test()
            .await
            .unwrap()
            .get(&ExecutionId::Derived(child_execution_id.clone()))
            .await
            .unwrap();
        assert!(child_log.pending_state.is_finished());
        assert_eq!(
            Version(2),
            child_log.next_version,
            "created = 0, locked = 1, with_single_result = 2"
        );
        assert_eq!(
            ExecutionRequest::Finished {
                retval: SupportedFunctionReturnValue::ExecutionFailure(expected_child_err),
                http_client_traces: None
            },
            child_log.last_event().event
        );
        let parent_log = db_pool
            .connection_test()
            .await
            .unwrap()
            .get(&parent_execution_id)
            .await
            .unwrap();
        assert_matches!(
            parent_log.pending_state,
            PendingState::PendingAt(PendingStatePendingAt {
                scheduled_at,
                last_lock: Some(LockedBy { executor_id: found_executor_id, run_id: _}),
            }) if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
            "parent should be back to pending"
        );
        let (found_join_set_id, found_child_execution_id, child_finished_version, found_result) = assert_matches!(
            parent_log.responses.last().map(|resp| &resp.event),
            Some(JoinSetResponseEventOuter{
                created_at: at,
                event: JoinSetResponseEvent{
                    join_set_id: found_join_set_id,
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: found_child_execution_id,
                        finished_version,
                        result: found_result,
                    }
                }
            })
             if *at == sim_clock.now()
            => (found_join_set_id, found_child_execution_id, finished_version, found_result)
        );
        assert_eq!(join_set_id, *found_join_set_id);
        assert_eq!(child_execution_id, *found_child_execution_id);
        assert_eq!(child_log.next_version, *child_finished_version);
        assert_matches!(
            found_result,
            SupportedFunctionReturnValue::ExecutionFailure(_)
        );

        db_close.close().await;
    }

    #[derive(Clone, Debug)]
    struct SleepyWorker {
        duration: Duration,
        result: SupportedFunctionReturnValue,
        exported: [FunctionMetadata; 1],
    }

    #[async_trait]
    impl Worker for SleepyWorker {
        async fn run(&self, ctx: WorkerContext) -> WorkerResult {
            tokio::time::sleep(self.duration).await;
            WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                retval: self.result.clone(),
                version: ctx.version,
                http_client_traces: None,
            }))
        }

        fn exported_functions_noext(&self) -> &[FunctionMetadata] {
            &self.exported
        }
    }

    #[rstest]
    #[tokio::test]
    async fn hanging_lock_should_be_cleaned_and_execution_retried(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let lock_expiry = Duration::from_millis(100);
        let timeout_duration = Duration::from_millis(300);
        let retry_config = ComponentRetryConfig {
            max_retries: Some(1),
            retry_exp_backoff: timeout_duration,
        };
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry,
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config,
            locking_strategy,
        };

        let worker = Arc::new(SleepyWorker {
            duration: lock_expiry + Duration::from_millis(1), // sleep more than allowed by the lock expiry
            result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
            exported: [FunctionMetadata {
                ffqn: FFQN_SOME,
                parameter_types: ParameterTypes::default(),
                return_type: RETURN_TYPE_DUMMY,
                extension: None,
                submittable: true,
            }],
        });
        // Create an execution
        let execution_id = ExecutionId::generate();
        let db_connection = db_pool.connection_test().await.unwrap();
        db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: execution_id.clone(),
                ffqn: FFQN_SOME,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();

        let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
        let (executor, _close_tx) = ExecTask::new_test(
            exec_config.clone(),
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );
        let db_exec = db_pool.db_exec_conn().await.unwrap();
        let mut first_execution_progress = executor
            .tick(
                db_exec.as_ref(),
                sim_clock.now(),
                RunId::generate(),
                DEPLOYMENT_ID_DUMMY,
            )
            .await
            .unwrap();
        assert_eq!(1, first_execution_progress.executions.len());
        // Started hanging, wait for lock expiry.
        sim_clock.move_time_forward(lock_expiry);
        // cleanup should be called
        let now_after_first_lock_expiry = sim_clock.now();
        {
            debug!(now = %now_after_first_lock_expiry, "Expecting an expired lock");
            let cleanup_progress = executor
                .tick(
                    db_pool.db_exec_conn().await.unwrap().as_ref(),
                    now_after_first_lock_expiry,
                    RunId::generate(),
                    DEPLOYMENT_ID_DUMMY,
                )
                .await
                .unwrap();
            assert!(cleanup_progress.executions.is_empty());
        }
        {
            let expired_locks = expired_timers_watcher::tick(
                db_pool.connection().await.unwrap().as_ref(),
                now_after_first_lock_expiry,
            )
            .await
            .unwrap()
            .expired_locks;
            assert_eq!(1, expired_locks);
        }
        assert!(
            !first_execution_progress
                .executions
                .pop()
                .unwrap()
                .1
                .is_finished()
        );

        let execution_log = db_connection.get(&execution_id).await.unwrap();
        let expected_first_timeout_expiry = now_after_first_lock_expiry + timeout_duration;
        assert_matches!(
            &execution_log.events.get(2).unwrap(),
            ExecutionEvent {
                event: ExecutionRequest::TemporarilyTimedOut { backoff_expires_at, .. },
                created_at: at,
                backtrace_id: None,
                version: Version(2),
            } if *at == now_after_first_lock_expiry && *backoff_expires_at == expected_first_timeout_expiry
        );
        assert_matches!(
            execution_log.pending_state,
            PendingState::PendingAt(PendingStatePendingAt {
                scheduled_at: found_scheduled_by,
                last_lock: Some(LockedBy {
                    executor_id: found_executor_id,
                    run_id: _,
                }),
            }) if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
        );
        sim_clock.move_time_forward(timeout_duration);
        let now_after_first_timeout = sim_clock.now();
        debug!(now = %now_after_first_timeout, "Second execution should hang again and result in a permanent timeout");

        let mut second_execution_progress = executor
            .tick(
                db_pool.db_exec_conn().await.unwrap().as_ref(),
                now_after_first_timeout,
                RunId::generate(),
                DEPLOYMENT_ID_DUMMY,
            )
            .await
            .unwrap();
        assert_eq!(1, second_execution_progress.executions.len());

        // Started hanging, wait for lock expiry.
        sim_clock.move_time_forward(lock_expiry);
        // cleanup should be called
        let now_after_second_lock_expiry = sim_clock.now();
        debug!(now = %now_after_second_lock_expiry, "Expecting the second lock to be expired");
        {
            let cleanup_progress = executor
                .tick(
                    db_pool.db_exec_conn().await.unwrap().as_ref(),
                    now_after_second_lock_expiry,
                    RunId::generate(),
                    DEPLOYMENT_ID_DUMMY,
                )
                .await
                .unwrap();
            assert!(cleanup_progress.executions.is_empty());
        }
        {
            let expired_locks = expired_timers_watcher::tick(
                db_pool.connection().await.unwrap().as_ref(),
                now_after_second_lock_expiry,
            )
            .await
            .unwrap()
            .expired_locks;
            assert_eq!(1, expired_locks);
        }
        assert!(
            !second_execution_progress
                .executions
                .pop()
                .unwrap()
                .1
                .is_finished()
        );

        drop(db_connection);
        drop(executor);
        db_close.close().await;
    }
}