obeli-sk-wasm-workers 0.30.0

use super::caching_db_connection::CacheableDbEvent;
use super::caching_db_connection::CachingDbConnection;
use super::deadline_tracker::DeadlineTracker;
use super::event_history::ProcessingStatus::Processed;
use super::event_history::ProcessingStatus::Unprocessed;
use super::host_exports::execution_id_derived_into_wast_val;
use super::host_exports::execution_id_into_wast_val;
use super::host_exports::v4_0_0::obelisk::types::execution::GetExtensionError;
use super::workflow_ctx::WorkflowFunctionError;
use super::workflow_worker::JoinNextBlockingStrategy;
use crate::activity::cancel_registry::CancelRegistry;
use crate::workflow::event_history::response_id::INVALID_CHILD_TYPE_FOR_DELAYS;
use crate::workflow::event_history::response_id::ResponseId;
use crate::workflow::host_exports::ffqn_into_wast_val;
use crate::workflow::host_exports::v4_0_0::obelisk::types::execution as types_execution;
use crate::workflow::host_exports::v4_0_0::obelisk::types::join_set as types_join_set;
use assert_matches::assert_matches;
use chrono::{DateTime, Utc};
use concepts::ComponentId;
use concepts::ComponentType;
use concepts::ExecutionMetadata;
use concepts::FunctionRegistry;
use concepts::InvalidNameError;
use concepts::JoinSetId;
use concepts::JoinSetKind;
use concepts::SupportedFunctionReturnValue;
use concepts::prefixed_ulid::DelayId;
use concepts::prefixed_ulid::ExecutionIdDerived;
use concepts::storage;
use concepts::storage::AppendEventsToExecution;
use concepts::storage::AppendResponseToExecution;
use concepts::storage::BacktraceInfo;
use concepts::storage::DbErrorGeneric;
use concepts::storage::DbErrorReadWithTimeout;
use concepts::storage::DbErrorWrite;
use concepts::storage::DbErrorWriteNonRetriable;
use concepts::storage::HistoryEventScheduleAt;
use concepts::storage::Locked;
use concepts::storage::PersistKind;
use concepts::storage::{
    AppendRequest, CreateRequest, ExecutionEventInner, JoinSetResponse, JoinSetResponseEvent,
    Version,
};
use concepts::storage::{HistoryEvent, JoinSetRequest};
use concepts::{ExecutionId, StrVariant};
use concepts::{FunctionFqn, Params};
use hashbrown::HashMap;
use indexmap::IndexMap;
use indexmap::indexmap;
use std::fmt::Debug;
use std::fmt::Display;
use std::sync::Arc;
use std::time::Duration;
use strum::IntoStaticStr;
use tracing::Level;
use tracing::Span;
use tracing::info;
use tracing::instrument;
use tracing::{debug, error, trace};
use val_json::wast_val::WastVal;
use wasmtime::component::Val;

#[derive(Debug)]
enum ChildReturnValue {
    WastVal(WastVal),
    JoinSetCreate(JoinSetId),
    JoinNext(Result<(ResponseId, Result<(), ()>), types_join_set::JoinNextError>),
    JoinNextRequestingFfqn(Result<(ExecutionIdDerived, WastVal), AwaitNextExtensionError>),
    OneOffDelay {
        scheduled_at: DateTime<Utc>,
        result: Result<(), ()>,
    },
    SubmitDelay,
}

#[derive(PartialEq, Eq, Clone, Copy, Debug)]
enum ProcessingStatus {
    Unprocessed,
    Processed,
}

#[derive(Debug, Clone, thiserror::Error)]
pub(crate) enum ApplyError {
    // fatal errors:
    #[error("nondeterminism detected: `{0}`")]
    NondeterminismDetected(String),
    // retriable errors:
    #[error("interrupt, db updated")]
    InterruptDbUpdated,
    #[error(transparent)]
    DbError(DbErrorWrite),
    #[error("constraint violation: {0}")]
    ConstraintViolation(StrVariant),
}

#[expect(clippy::struct_field_names)]
pub(crate) struct EventHistory {
    join_next_blocking_strategy: JoinNextBlockingStrategy,
    // Contains requests (events produced by the workflow)
    event_history: Vec<(HistoryEvent, ProcessingStatus, Version)>,
    // Used for `-get`ting the processed response by Execution Id.
    index_child_exe_to_processed_response_idx: HashMap<ExecutionIdDerived, usize>,
    index_child_exe_to_ffqn: HashMap<ExecutionIdDerived, FunctionFqn>,
    // Used for closing join sets in reverse order. One-off join sets are ignored.

    // index_join_set_to_created_child_requests:
    // IndexMap<JoinSetId, usize /* created child count */>,
    index_delay_id_to_expires_at: IndexMap<DelayId, DateTime<Utc>>,

    // Used for closing join sets.
    // close_requests: std::collections::HashMap<JoinSetId, usize>,
    // closed_join_sets: HashSet<JoinSetId>,
    responses: Vec<(JoinSetResponseEvent, ProcessingStatus)>,
    worker_span: Span,
    deadline_tracker: Box<dyn DeadlineTracker>,
    lock_extension: Duration,
    locked_event: Locked,
    fn_registry: Arc<dyn FunctionRegistry>,
    cancel_registry: CancelRegistry,

    // Tracks join set contents for closing. One-offs are ignored, response id is removed when the response is processed.
    index_join_set_to_unawaited_requests: IndexMap<JoinSetId, IndexMap<ResponseId, ComponentType>>,
}

#[derive(Debug)]
enum FindMatchingResponse {
    Found(ChildReturnValue),
    NotFound,
    FoundRequestButNotResponse,
}

impl EventHistory {
    #[expect(clippy::too_many_arguments)]
    pub(crate) fn new(
        event_history: Vec<(HistoryEvent, Version)>,
        responses: Vec<JoinSetResponseEvent>,
        join_next_blocking_strategy: JoinNextBlockingStrategy,
        fn_registry: Arc<dyn FunctionRegistry>,
        cancel_registry: CancelRegistry,
        deadline_tracker: Box<dyn DeadlineTracker>,
        locked_event: Locked,
        lock_extension: Duration,
        worker_span: Span,
    ) -> EventHistory {
        EventHistory {
            index_child_exe_to_processed_response_idx: HashMap::default(),
            index_child_exe_to_ffqn: HashMap::default(),
            index_delay_id_to_expires_at: IndexMap::default(),
            event_history: event_history
                .into_iter()
                .map(|(event, version)| (event, Unprocessed, version))
                .collect(),
            responses: responses
                .into_iter()
                .map(|event| (event, Unprocessed))
                .collect(),
            join_next_blocking_strategy,
            worker_span,
            deadline_tracker,
            fn_registry,
            cancel_registry,
            locked_event,
            lock_extension,
            index_join_set_to_unawaited_requests: IndexMap::default(),
        }
    }

    /// Unprocessed requests imply replaying
    pub(crate) fn has_unprocessed_requests(&self) -> bool {
        self.first_unprocessed_request().is_some()
    }

    pub(crate) fn join_set_name_exists(&self, join_set_name: &str, kind: JoinSetKind) -> bool {
        // TODO: optimize
        self.event_history
            .iter()
            .any(|(event, processing_status, _version)|
            // Do not look into the future as it would break replay.
            *processing_status == ProcessingStatus::Processed &&
            matches!(event, HistoryEvent::JoinSetCreate { join_set_id: found, .. }
                if found.name.as_ref() == join_set_name && found.kind == kind))
    }

    pub(crate) fn join_set_count(&self, kind: JoinSetKind) -> usize {
        // TODO: optimize
        self.event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                // Do not look into the future as it would break replay.
                *processing_status == ProcessingStatus::Processed
                    && matches!(
                        event,
                        HistoryEvent::JoinSetCreate {
                            join_set_id: JoinSetId {
                                kind: found_kind,
                                ..
                            },
                            ..
                        }
                        if *found_kind == kind
                    )
            })
            .count()
    }

    pub(crate) fn execution_count(&self, join_set_id: &JoinSetId) -> usize {
        // TODO: optimize
        self.event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                // Do not look into the future as it would break replay.
                *processing_status == ProcessingStatus::Processed
                    && matches!(
                        event,
                        HistoryEvent::JoinSetRequest {
                            join_set_id: found,
                            request: JoinSetRequest::ChildExecutionRequest { .. },
                        }
                        if found == join_set_id
                    )
            })
            .count()
    }

    /// Returns `WorkflowFunctionError` that is interpreted for interrupt etc. by the worker.
    async fn apply(
        &mut self,
        event_call: EventCall,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, WorkflowFunctionError> {
        Ok(self
            .apply_inner(event_call, db_connection, called_at)
            .await?)
    }

    /// Apply the event and wait if new, replay if already in the event history, or
    /// apply with an interrupt.
    #[instrument(skip_all, fields(otel.name = format!("apply {event_call}"), ?event_call))]
    async fn apply_inner(
        &mut self,
        event_call: EventCall,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, ApplyError> {
        debug!("applying {event_call:?}, Version:{}", db_connection.version);

        if self.deadline_tracker.close_to_expired() && self.lock_extension > Duration::ZERO {
            self.extend_lock(db_connection, called_at)
                .await
                .map_err(ApplyError::DbError)?;
        }

        if let Some(resp) = self.find_matching_atomic(&event_call)? {
            trace!("found_atomic: {resp:?}");
            return Ok(resp);
        }

        match event_call {
            EventCall::NonBlocking(event_call) => {
                // Events that cannot block waiting for response.
                let cloned_non_blocking = event_call.clone();
                let (event, version) = self
                    .append_to_db_non_blocking(event_call, db_connection, called_at)
                    .await
                    .map_err(ApplyError::DbError)?;
                self.event_history.push((event, Unprocessed, version));
                trace!("find_matching_atomic must mark the non-blocking event as Processed");
                let non_blocking_resp = self
                    .find_matching_atomic(&EventCall::NonBlocking(cloned_non_blocking))?
                    .expect("just stored the event as Unprocessed, it must be found");
                Ok(non_blocking_resp)
            }
            EventCall::Blocking(event_call) => {
                let lock_expires_at =
                    if self.join_next_blocking_strategy == JoinNextBlockingStrategy::Interrupt {
                        called_at
                    } else {
                        self.locked_event.lock_expires_at
                    };
                self.apply_blocking(event_call, db_connection, lock_expires_at, called_at)
                    .await
            }
        }
    }

    async fn extend_lock(
        &mut self,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), DbErrorWrite> {
        self.locked_event.lock_expires_at = self.deadline_tracker.extend_by(self.lock_extension);
        let append_req = AppendRequest {
            created_at: called_at,
            event: ExecutionEventInner::Locked(self.locked_event.clone()),
        };
        info!(
            "Extending the lock at version {version}",
            version = db_connection.version
        );
        db_connection
            .append_blocking(
                db_connection.execution_id.clone(),
                append_req,
                called_at,
                None,
                &self.locked_event.component_id,
            )
            .await?;
        Ok(())
    }

    async fn apply_blocking(
        &mut self,
        event_call: EventCallBlocking,
        db_connection: &mut CachingDbConnection,
        lock_expires_at: DateTime<Utc>,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, ApplyError> {
        let join_next_variant = event_call.join_next_variant();
        let keys = event_call.as_keys();
        // Create and append HistoryEvents.
        let history_events = self
            .append_to_db_blocking(event_call, db_connection, called_at, lock_expires_at)
            .await
            .map_err(ApplyError::DbError)?;
        assert!(
            !history_events.is_empty(),
            "each EventCall must produce at least one HistoryEvent"
        );
        // Extend event history.
        self.event_history.extend(
            history_events
                .into_iter()
                .map(|(event, version)| (event, Unprocessed, version)),
        );

        let last_key_idx = keys.len() - 1;
        for (idx, key) in keys.into_iter().enumerate() {
            let res = self.process_event_by_key(&key)?;
            if idx == last_key_idx
                && let FindMatchingResponse::Found(res) = res
            {
                // Last key was marked as processed.
                assert_eq!(
                    Processed,
                    self.event_history
                        .last()
                        .expect("checked that `history_events` is not empty")
                        .1
                );
                return Ok(res);
            }
        }
        // Now either wait or interrupt.
        // TODO: perf: if start_from_index was at top, it should move forward to n - 1

        if matches!(
            self.join_next_blocking_strategy,
            JoinNextBlockingStrategy::Await { .. }
        ) {
            // JoinNext was written, wait for next response.
            debug!(join_set_id = %join_next_variant.join_set_id(), "Waiting for {join_next_variant:?}");
            let key = join_next_variant.as_key();

            // Subscribe to the next response.
            while let Some(timeout_fut) = self.deadline_tracker.track() {
                let next_responses = match db_connection
                    .subscribe_to_next_responses(
                        &db_connection.execution_id,
                        self.responses.len(),
                        timeout_fut,
                    )
                    .await
                {
                    Ok(ok) => ok,
                    Err(DbErrorReadWithTimeout::DbErrorRead(err)) => {
                        return Err(ApplyError::DbError(DbErrorWrite::from(err)));
                    }
                    Err(DbErrorReadWithTimeout::Timeout) => {
                        info!("Giving up on waiting for response");
                        return Err(ApplyError::InterruptDbUpdated);
                    }
                };
                debug!("Got next responses {next_responses:?}");
                self.responses.extend(
                    next_responses
                        .into_iter()
                        .map(|outer| (outer.event, Unprocessed)),
                );
                trace!("All responses: {:?}", self.responses);
                if let FindMatchingResponse::Found(accept_resp) = self.process_event_by_key(&key)? {
                    debug!(join_set_id = %join_next_variant.join_set_id(), "Got result");
                    return Ok(accept_resp);
                }
            }
            Err(ApplyError::InterruptDbUpdated)
        } else {
            debug!(join_set_id = %join_next_variant.join_set_id(),  "Interrupting on {join_next_variant:?}");
            Err(ApplyError::InterruptDbUpdated)
        }
    }

    // Transforms `join_set_close_inner` error to `WorkflowFunctionError` so
    // that the interrupt can be properly interpreted.
    pub(crate) async fn join_set_close(
        &mut self,
        join_set_id: &JoinSetId,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<(), WorkflowFunctionError> {
        self.join_set_close_inner(join_set_id, db_connection, called_at, wasm_backtrace)
            .await
            .map_err(WorkflowFunctionError::from)
    }
    /// Deterministically cancel activites and delays, emit `JoinNext` and wait for the response.
    async fn join_set_close_inner(
        &mut self,
        join_set_id: &JoinSetId,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<(), ApplyError> {
        let (_, response_ids) = self
            .index_join_set_to_unawaited_requests
            .shift_remove_entry(join_set_id)
            .ok_or_else(|| {
                ApplyError::ConstraintViolation(
                    format!("not found in open join sets: `{join_set_id}`").into(),
                )
            })?;
        debug!("Closing `{join_set_id}` with {response_ids:?}");

        let join_next_count = response_ids.len();
        // Attempt to cancel activities and delays.
        for (response_id, component_type) in response_ids.iter().rev() {
            if let ResponseId::ChildExecutionId(child_execution_id_derived) = response_id
                && component_type.is_activity()
            {
                let res = self
                    .cancel_registry
                    .cancel(
                        db_connection.db_connection.as_ref(),
                        &ExecutionId::Derived(child_execution_id_derived.clone()),
                        called_at,
                    )
                    .await;
                if let Err(err) = res {
                    debug!("Ignoring failure to cancel {child_execution_id_derived} - {err:?}");
                }
            } else if let ResponseId::DelayId(delay_id) = response_id {
                debug!("Cancelling {delay_id}");
                let res = storage::cancel_delay(
                    db_connection.db_connection.as_ref(),
                    delay_id.clone(),
                    called_at,
                )
                .await;
                if let Err(err) = res {
                    // This means that the watcher expired the delay in the mean time.
                    trace!("Ignoring failure to cancel {delay_id} - {err:?}");
                }
            } // else child workflow, no cancellation
        }

        for _ in 0..join_next_count {
            self.apply_inner(
                EventCall::Blocking(EventCallBlocking::JoinNext(JoinNext {
                    join_set_id: join_set_id.clone(),
                    closing: true,
                    wasm_backtrace: wasm_backtrace.clone(),
                })),
                db_connection,
                called_at,
            )
            .await?;
        }
        Ok(())
    }

    /// Final determinism check: there must be no unprocessed requests, otherwise the
    /// code is trimming previous run.
    #[instrument(skip_all)]
    pub(crate) async fn finalize(
        &mut self,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), ApplyError> {
        while let Some(join_set_id) = self
            .index_join_set_to_unawaited_requests
            .iter()
            .rev()
            .find_map(|(js, remaining)| {
                if !remaining.is_empty() {
                    Some(js.clone())
                } else {
                    None
                }
            })
        {
            self.join_set_close_inner(&join_set_id, db_connection, called_at, None)
                .await?;
        }
        if let Some((_found_idx, first_unprocessed, version)) = self.first_unprocessed_request() {
            return Err(ApplyError::NondeterminismDetected(format!(
                "found unprocessed request stored at version {version}: event: {first_unprocessed}",
            )));
        }
        Ok(())
    }

    // Return ChildReturnValue if response is found
    fn find_matching_atomic(
        &mut self,
        event_call: &EventCall,
    ) -> Result<Option<ChildReturnValue>, ApplyError> {
        let keys = event_call.as_keys();
        assert!(!keys.is_empty());
        let last_key_idx = keys.len() - 1;
        for (idx, key) in keys.into_iter().enumerate() {
            let resp = self.process_event_by_key(&key)?;
            match resp {
                FindMatchingResponse::NotFound => {
                    assert_eq!(idx, 0, "NotFound must be returned on the first key");
                    return Ok(None);
                }
                FindMatchingResponse::FoundRequestButNotResponse => {
                    unreachable!(
                        "FoundRequestButNotResponse in find_matching_atomic - {event_call:?}"
                    );
                }
                FindMatchingResponse::Found(found) => {
                    if idx == last_key_idx {
                        return Ok(Some(found));
                    }
                }
            }
        }
        unreachable!()
    }

    fn first_unprocessed_request(&self) -> Option<(usize, &HistoryEvent, &Version)> {
        self.event_history
            .iter()
            .enumerate()
            .find_map(|(idx, (event, status, version))| {
                if *status == Unprocessed {
                    Some((idx, event, version))
                } else {
                    None
                }
            })
    }

    fn mark_next_unprocessed_response(
        &'_ mut self,
        parent_event_idx: usize, // needs to be marked as Processed as well if found
        join_set_id: &JoinSetId,
    ) -> Option<JoinSetResponseEnriched<'_>> {
        trace!(
            "mark_next_unprocessed_response responses: {:?}",
            self.responses
        );
        let found_resp_idx =
            self.responses
                .iter()
                .enumerate()
                .find_map(|(idx, (event, status))| {
                    if *status == Unprocessed
                        && let JoinSetResponseEvent {
                            join_set_id: found_join_set_id,
                            event: _,
                        } = event
                        && found_join_set_id == join_set_id
                    {
                        Some(idx)
                    } else {
                        None
                    }
                });
        if let Some(found_resp_idx) = found_resp_idx {
            self.event_history[parent_event_idx].1 = Processed;
            self.responses[found_resp_idx].1 = Processed;
            let enriched = {
                match &self.responses[found_resp_idx].0.event {
                    JoinSetResponse::ChildExecutionFinished {
                        child_execution_id,
                        finished_version: _,
                        result,
                    } => {
                        self.index_child_exe_to_processed_response_idx
                            .insert(child_execution_id.clone(), found_resp_idx);

                        let response_ffqn = self
                            .index_child_exe_to_ffqn
                            .get(child_execution_id)
                            .expect("if finished the index must have it");
                        JoinSetResponseEnriched::ChildExecutionFinished(ChildExecutionFinished {
                            child_execution_id,
                            result,
                            response_ffqn,
                        })
                    }
                    JoinSetResponse::DelayFinished { delay_id, result } => {
                        // Find matching DelayRequest to extract expires_at
                        let expires_at = *self
                            .index_delay_id_to_expires_at
                            .get(delay_id)
                            .expect("found delay-id must have been indexed");
                        JoinSetResponseEnriched::DelayFinished {
                            delay_id,
                            expires_at,
                            result: *result,
                        }
                    }
                }
            };
            Some(enriched)
        } else {
            None
        }
    }

    fn process_event_by_key(
        &mut self,
        key: &DeterministicKey,
    ) -> Result<FindMatchingResponse, ApplyError> {
        let Some((found_idx, found_request_event, _version)) = self.first_unprocessed_request()
        else {
            return Ok(FindMatchingResponse::NotFound);
        };
        trace!("Finding match for {key:?}, [{found_idx}] {found_request_event:?}");
        match (key, found_request_event) {
            (
                DeterministicKey::CreateJoinSet { join_set_id },
                HistoryEvent::JoinSetCreate {
                    join_set_id: found_join_set_id,
                },
            ) if *join_set_id == *found_join_set_id => {
                trace!(%join_set_id, "Matched JoinSet");
                self.event_history[found_idx].1 = Processed;
                // if this is a [`EventCall::CreateJoinSet`] , return join set id
                Ok(FindMatchingResponse::Found(
                    ChildReturnValue::JoinSetCreate(join_set_id.clone()),
                ))
            }

            (
                DeterministicKey::Persist { value, kind },
                HistoryEvent::Persist {
                    value: found_value,
                    kind: found_kind,
                },
            ) if *value == *found_value && *kind == *found_kind => {
                trace!("Matched Persist");
                self.event_history[found_idx].1 = Processed;
                // if this is a [`EventCall::CreateJoinSet`] , return join set id
                Ok(FindMatchingResponse::Found(ChildReturnValue::WastVal(
                    match kind {
                        PersistKind::RandomString { .. } => {
                            WastVal::String(String::from_utf8(value.clone()).map_err(|err| {
                                error!("Persisted string must be UTF-8 - {err:?}");
                                ApplyError::DbError(DbErrorWrite::from(
                                    DbErrorGeneric::Uncategorized(StrVariant::from(format!(
                                        "persisted string must be UTF-8 - {err:?}"
                                    ))),
                                ))
                            })?)
                        }
                        PersistKind::RandomU64 { .. } => {
                            if value.len() != 8 {
                                return Err(ApplyError::DbError(DbErrorWrite::from(
                                    DbErrorGeneric::Uncategorized(
                                        "value cannot be deserialized to u64".into(),
                                    ),
                                )));
                            }
                            let value: [u8; 8] = value[..8].try_into().expect("size checked above");
                            let value = storage::from_bytes_to_u64(value);
                            WastVal::U64(value)
                        }
                    },
                )))
            }

            (
                DeterministicKey::ChildExecutionRequest {
                    join_set_id,
                    child_execution_id: execution_id,
                    target_ffqn,
                    params,
                },
                HistoryEvent::JoinSetRequest {
                    join_set_id: found_join_set_id,
                    request:
                        JoinSetRequest::ChildExecutionRequest {
                            child_execution_id,
                            target_ffqn: stored_target_ffqn,
                            params: stored_params,
                        },
                },
            ) if *join_set_id == *found_join_set_id
                && *execution_id == *child_execution_id
                && target_ffqn == stored_target_ffqn
                && params == stored_params =>
            {
                trace!(%child_execution_id, %join_set_id, "Matched JoinSetRequest::ChildExecutionRequest");
                self.index_child_exe_to_ffqn
                    .insert(child_execution_id.clone(), target_ffqn.clone());
                self.event_history[found_idx].1 = Processed;
                // if this is a [`EventCall::StartAsync`] , return execution id
                Ok(FindMatchingResponse::Found(ChildReturnValue::WastVal(
                    execution_id_derived_into_wast_val(execution_id),
                )))
            }

            (
                DeterministicKey::DelayRequest {
                    join_set_id,
                    delay_id,
                    schedule_at,
                },
                HistoryEvent::JoinSetRequest {
                    join_set_id: found_join_set_id,
                    request:
                        JoinSetRequest::DelayRequest {
                            delay_id: found_delay_id,
                            expires_at,
                            schedule_at: found_schedule_at,
                        },
                },
            ) if *join_set_id == *found_join_set_id
                && *delay_id == *found_delay_id
                && schedule_at == found_schedule_at =>
            {
                trace!(%delay_id, %join_set_id, "Matched JoinSetRequest::DelayRequest");
                self.index_delay_id_to_expires_at
                    .insert(delay_id.clone(), *expires_at);
                self.event_history[found_idx].1 = Processed;
                // return delay id
                Ok(FindMatchingResponse::Found(ChildReturnValue::SubmitDelay))
            }

            (
                DeterministicKey::JoinNextChild {
                    join_set_id,
                    kind: JoinNextChildKind::AwaitNext,
                    requested_ffqn,
                },
                HistoryEvent::JoinNextTooMany {
                    join_set_id: found_join_set_id,
                    requested_ffqn: found_requested_ffqn,
                },
            ) if *join_set_id == *found_join_set_id
                && Some(requested_ffqn) == found_requested_ffqn.as_ref() =>
            {
                trace!(%join_set_id, "matched JoinNextChild with JoinNextTooMany");
                self.event_history[found_idx].1 = Processed;

                Ok(FindMatchingResponse::Found(
                    ChildReturnValue::JoinNextRequestingFfqn(Err(
                        AwaitNextExtensionError::AllProcessed,
                    )),
                ))
            }

            (
                DeterministicKey::JoinNextChild {
                    join_set_id,
                    kind,
                    requested_ffqn,
                },
                HistoryEvent::JoinNext {
                    join_set_id: found_join_set_id,
                    requested_ffqn: Some(found_requested_ffqn),
                    run_expires_at: _,
                    closing: false, // JoinNextChild originates from user's code
                },
            ) if *join_set_id == *found_join_set_id && requested_ffqn == found_requested_ffqn => {
                trace!(%join_set_id, "Peeked at JoinNext - Child");
                match self.mark_next_unprocessed_response(found_idx, join_set_id) {
                    Some(JoinSetResponseEnriched::ChildExecutionFinished(
                        ChildExecutionFinished {
                            child_execution_id,
                            result,
                            response_ffqn,
                        },
                    )) if requested_ffqn == response_ffqn => {
                        trace!(%join_set_id, "Matched JoinNext & ChildExecutionFinished");
                        let response_ffqn = response_ffqn.clone();
                        let child_execution_id = child_execution_id.clone();
                        let inner_res = result.clone().into_wast_val( || self.fn_registry.get_ret_type(&response_ffqn)
                                        .expect("response_ffqn can only be exported and no-ext, thus must be returned by get_ret_type"));
                        match kind {
                            JoinNextChildKind::DirectCall => Ok(FindMatchingResponse::Found(
                                ChildReturnValue::WastVal(inner_res),
                            )),
                            JoinNextChildKind::AwaitNext => {
                                // result<(execution-id, inner-res>, await-next-extension-error>
                                Ok(FindMatchingResponse::Found(
                                    ChildReturnValue::JoinNextRequestingFfqn(Ok((
                                        child_execution_id,
                                        inner_res,
                                    ))),
                                ))
                            }
                        }
                    }
                    Some(JoinSetResponseEnriched::ChildExecutionFinished(
                        ChildExecutionFinished {
                            child_execution_id,
                            result: _,
                            response_ffqn, // mismatch between ffqns
                        },
                    )) => {
                        let function_mismatch = AwaitNextExtensionError::FunctionMismatch {
                            specified_function: requested_ffqn.clone(),
                            actual_function: Some(response_ffqn.clone()),
                            actual_id: ResponseId::ChildExecutionId(child_execution_id.clone()),
                        };
                        Ok(FindMatchingResponse::Found(
                            ChildReturnValue::JoinNextRequestingFfqn(Err(function_mismatch)),
                        ))
                    }
                    Some(JoinSetResponseEnriched::DelayFinished {
                        delay_id,
                        result: _,
                        expires_at: _,
                    }) => {
                        let function_mismatch = AwaitNextExtensionError::FunctionMismatch {
                            specified_function: requested_ffqn.clone(),
                            actual_function: None,
                            actual_id: ResponseId::DelayId(delay_id.clone()),
                        };
                        Ok(FindMatchingResponse::Found(
                            ChildReturnValue::JoinNextRequestingFfqn(Err(function_mismatch)),
                        ))
                    }
                    None => Ok(FindMatchingResponse::FoundRequestButNotResponse), // no progress, still at JoinNext
                }
            }

            (
                DeterministicKey::JoinNextDelay { join_set_id },
                HistoryEvent::JoinNext {
                    join_set_id: found_join_set_id,
                    requested_ffqn: None,
                    closing: false, // JoinNextDelay originates from user's code
                    run_expires_at: _,
                },
            ) if *join_set_id == *found_join_set_id => {
                trace!(
                    %join_set_id, "Peeked at JoinNext - Delay");
                match self.mark_next_unprocessed_response(found_idx, join_set_id) {
                    Some(JoinSetResponseEnriched::DelayFinished {
                        expires_at: scheduled_at,
                        result,
                        delay_id: _, // one-shot
                    }) => {
                        trace!(%join_set_id, "Matched JoinNext & DelayFinished");
                        Ok(FindMatchingResponse::Found(ChildReturnValue::OneOffDelay {
                            scheduled_at,
                            result,
                        }))
                    }
                    None => Ok(FindMatchingResponse::FoundRequestButNotResponse), // no progress, still at JoinNext
                    Some(JoinSetResponseEnriched::ChildExecutionFinished { .. }) => unreachable!(
                        "DeterministicKey::JoinNextDelay is emitted only on one-shot join sets"
                    ),
                }
            }

            (
                DeterministicKey::JoinNext {
                    join_set_id,
                    closing,
                },
                HistoryEvent::JoinNext {
                    join_set_id: found_join_set_id,
                    requested_ffqn: None, // closing and `join-next` are agnostic of ffqn.
                    run_expires_at: _,
                    closing: found_closing,
                },
            ) if *join_set_id == *found_join_set_id && closing == found_closing => {
                trace!(%join_set_id, "DeterministicKey::JoinNext(closing:{closing}): Peeked at JoinNext");
                match self.mark_next_unprocessed_response(found_idx, join_set_id) {
                    Some(JoinSetResponseEnriched::ChildExecutionFinished(
                        ChildExecutionFinished {
                            child_execution_id,
                            result: res,
                            response_ffqn: _,
                        },
                    )) => {
                        trace!(%join_set_id, %child_execution_id, "DeterministicKey::JoinNext: Matched ChildExecutionFinished");
                        Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNext(Ok(
                            (
                                ResponseId::ChildExecutionId(child_execution_id.clone()),
                                res.as_pending_state_finished_result()
                                    .as_result()
                                    .map_err(|_| ()),
                            ),
                        ))))
                    }
                    Some(JoinSetResponseEnriched::DelayFinished {
                        delay_id,
                        result,
                        expires_at: _,
                    }) => {
                        trace!(%join_set_id, %delay_id, "DeterministicKey::JoinNext: Matched DelayFinished");
                        Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNext(Ok(
                            (ResponseId::DelayId(delay_id.clone()), result),
                        ))))
                    }
                    None => Ok(FindMatchingResponse::FoundRequestButNotResponse), // no progress, still at JoinNext
                }
            }

            (
                DeterministicKey::JoinNext {
                    join_set_id,
                    closing: false, // Runtime should never issue a bogus `JoinNext`.
                },
                HistoryEvent::JoinNextTooMany {
                    join_set_id: found_join_set_id,
                    requested_ffqn: None, // `join-next` is agnostic of ffqn.
                },
            ) if *join_set_id == *found_join_set_id => {
                trace!(%join_set_id, "matched JoinNext with JoinNextTooMany");
                self.event_history[found_idx].1 = Processed;
                Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNext(
                    Err(types_join_set::JoinNextError::AllProcessed),
                )))
            }

            (
                DeterministicKey::Schedule {
                    target_execution_id,
                    schedule_at,
                },
                HistoryEvent::Schedule {
                    execution_id: found_execution_id,
                    schedule_at: found_schedule_at,
                    ..
                },
            ) if *target_execution_id == *found_execution_id
                && schedule_at == found_schedule_at =>
            {
                trace!(%target_execution_id, "Matched Schedule");
                self.event_history[found_idx].1 = Processed;
                // return execution id
                Ok(FindMatchingResponse::Found(ChildReturnValue::WastVal(
                    execution_id_into_wast_val(target_execution_id),
                )))
            }

            (
                DeterministicKey::Stub {
                    target_execution_id,
                    return_value,
                },
                HistoryEvent::Stub {
                    target_execution_id: found_execution_id,
                    result: found_result,
                    persist_result: target_result,
                },
            ) if target_execution_id == found_execution_id && return_value == found_result => {
                trace!(%target_execution_id, "Matched Stub");
                let wat_val = match target_result {
                    Ok(()) => WastVal::Result(Ok(None)),
                    Err(()) => WastVal::Result(Err(Some(Box::new(WastVal::Variant(
                        "conflict".to_string(),
                        None,
                    ))))),
                };
                self.event_history[found_idx].1 = Processed;
                Ok(FindMatchingResponse::Found(ChildReturnValue::WastVal(
                    wat_val,
                )))
            }

            (key, found) => {
                let version = &self.event_history[found_idx].2;
                Err(ApplyError::NondeterminismDetected(format!(
                    "key does not match event stored at version {version}: key: {key}, event: {found}",
                )))
            }
        }
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_to_db_non_blocking(
        &mut self,
        event_call: EventCallNonBlocking,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(HistoryEvent, Version), DbErrorWrite> {
        trace!("append_to_db_non_blocking {}", db_connection.version);
        match event_call {
            EventCallNonBlocking::JoinSetCreate(JoinSetCreate {
                join_set_id,
                wasm_backtrace,
            }) => {
                // Cacheable event.
                debug!(%join_set_id, "CreateJoinSet: Creating new JoinSet");
                let event = HistoryEvent::JoinSetCreate { join_set_id };
                let history_event = (event.clone(), db_connection.version.clone());
                let join_set_create = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::JoinSetCreate {
                    request: join_set_create,
                    version: db_connection.version.clone(),
                    backtrace: wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
                        execution_id: db_connection.execution_id.clone(),
                        component_id: self.locked_event.component_id.clone(),
                        wasm_backtrace,
                        version_min_including: db_connection.version.clone(),
                        version_max_excluding: Version::new(db_connection.version.0 + 1),
                    }),
                };
                db_connection
                    .append_non_blocking(cacheable_event, called_at)
                    .await?;
                Ok(history_event)
            }

            EventCallNonBlocking::Persist(Persist {
                value,
                kind,
                wasm_backtrace,
            }) => {
                // Cacheable event.
                let event = HistoryEvent::Persist { value, kind };
                let history_event = (event.clone(), db_connection.version.clone());
                let request = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::Persist {
                    request,
                    version: db_connection.version.clone(),
                    backtrace: wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
                        execution_id: db_connection.execution_id.clone(),
                        component_id: self.locked_event.component_id.clone(),
                        wasm_backtrace,
                        version_min_including: db_connection.version.clone(),
                        version_max_excluding: Version::new(db_connection.version.0 + 1),
                    }),
                };
                db_connection
                    .append_non_blocking(cacheable_event, called_at)
                    .await?;
                Ok(history_event)
            }

            EventCallNonBlocking::SubmitChildExecution(SubmitChildExecution {
                target_ffqn,
                fn_component_id,
                join_set_id,
                child_execution_id,
                params,
                wasm_backtrace,
            }) => {
                // Cacheable event.
                debug!(%child_execution_id, %join_set_id, "StartAsync: appending ChildExecutionRequest");
                let event = HistoryEvent::JoinSetRequest {
                    join_set_id: join_set_id.clone(),
                    request: JoinSetRequest::ChildExecutionRequest {
                        child_execution_id: child_execution_id.clone(),
                        target_ffqn: target_ffqn.clone(),
                        params: params.clone(),
                    },
                };
                let history_event = (event.clone(), db_connection.version.clone());
                let append_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let child_req = CreateRequest {
                    created_at: called_at,
                    execution_id: ExecutionId::Derived(child_execution_id),
                    ffqn: target_ffqn,
                    params,
                    parent: Some((db_connection.execution_id.clone(), join_set_id)),
                    metadata: ExecutionMetadata::from_parent_span(&self.worker_span),
                    scheduled_at: called_at,
                    component_id: fn_component_id,
                    scheduled_by: None,
                };
                let cacheable_event = CacheableDbEvent::SubmitChildExecution {
                    request: append_req,
                    version: db_connection.version.clone(),
                    child_req,
                    backtrace: wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
                        execution_id: db_connection.execution_id.clone(),
                        component_id: self.locked_event.component_id.clone(),
                        wasm_backtrace,
                        version_min_including: db_connection.version.clone(),
                        version_max_excluding: Version::new(db_connection.version.0 + 1),
                    }),
                };

                db_connection
                    .append_non_blocking(cacheable_event, called_at)
                    .await?;
                Ok(history_event)
            }

            EventCallNonBlocking::SubmitDelay(SubmitDelay {
                join_set_id,
                delay_id,
                schedule_at,
                expires_at_if_new,
                wasm_backtrace,
            }) => {
                // Cacheable event.
                debug!(%delay_id, %join_set_id, "SubmitDelay");

                let event = HistoryEvent::JoinSetRequest {
                    join_set_id: join_set_id.clone(),
                    request: JoinSetRequest::DelayRequest {
                        delay_id,
                        expires_at: expires_at_if_new,
                        schedule_at,
                    },
                };
                let history_event = (event.clone(), db_connection.version.clone());
                let delay_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::SubmitDelay {
                    request: delay_req,
                    version: db_connection.version.clone(),
                    backtrace: wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
                        execution_id: db_connection.execution_id.clone(),
                        component_id: self.locked_event.component_id.clone(),
                        wasm_backtrace,
                        version_min_including: db_connection.version.clone(),
                        version_max_excluding: Version::new(db_connection.version.0 + 1),
                    }),
                };
                db_connection
                    .append_non_blocking(cacheable_event, called_at)
                    .await?;
                Ok(history_event)
            }

            EventCallNonBlocking::Schedule(Schedule {
                schedule_at,
                scheduled_at_if_new,
                execution_id: new_execution_id,
                ffqn,
                fn_component_id,
                params,
                wasm_backtrace,
            }) => {
                // Cacheable event.
                let event = HistoryEvent::Schedule {
                    execution_id: new_execution_id.clone(),
                    schedule_at,
                };

                let history_event = (event.clone(), db_connection.version.clone());
                let append_req = AppendRequest {
                    event: ExecutionEventInner::HistoryEvent { event },
                    created_at: called_at,
                };
                debug!(%new_execution_id, "ScheduleRequest: appending");
                let child_req = CreateRequest {
                    created_at: called_at,
                    execution_id: new_execution_id,
                    metadata: ExecutionMetadata::from_linked_span(&self.worker_span),
                    ffqn,
                    params,
                    parent: None, // Schedule breaks from the parent-child relationship to avoid a linked list
                    scheduled_at: scheduled_at_if_new,
                    component_id: fn_component_id,
                    scheduled_by: Some(db_connection.execution_id.clone()),
                };
                let non_blocking_event = CacheableDbEvent::Schedule {
                    request: append_req,
                    version: db_connection.version.clone(),
                    child_req,
                    backtrace: wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
                        execution_id: db_connection.execution_id.clone(),
                        component_id: self.locked_event.component_id.clone(),
                        wasm_backtrace,
                        version_min_including: db_connection.version.clone(),
                        version_max_excluding: Version::new(db_connection.version.0 + 1),
                    }),
                };

                db_connection
                    .append_non_blocking(non_blocking_event, called_at)
                    .await?;

                Ok(history_event)
            }

            EventCallNonBlocking::Stub(Stub {
                target_ffqn,
                target_execution_id,
                parent_id,
                join_set_id,
                result,
                wasm_backtrace,
            }) => {
                // Cannot be cacheable.
                // Attempt to write to target_execution_id, will continue on conflict.
                // Non-cacheable event. (could be turned into one)
                // The idempotent write is needed to avoid race with stub requests originating from gRPC.
                debug!(%target_execution_id, "StubRequest: Flushing and appending");

                // Flush the cache before getting the stub's create request, because it might be this execution's child.
                db_connection
                    .flush_non_blocking_event_cache(called_at)
                    .await?;
                // Check that the execution exists and FFQN matches.
                if target_ffqn
                    != db_connection
                        .get_create_request(&ExecutionId::Derived(target_execution_id.clone()))
                        .await?
                        .ffqn
                {
                    return Err(DbErrorWrite::NonRetriable(
                        DbErrorWriteNonRetriable::ValidationFailed("ffqn mismatch".into()),
                    ));
                }
                let stub_finished_version = Version::new(1); // Stub activities have no execution log except Created event.
                // Attempt to write to target_execution_id and its parent, ignoring the possible conflict error on this tx
                let write_attempt = {
                    let finished_req = AppendRequest {
                        created_at: called_at,
                        event: ExecutionEventInner::Finished {
                            result: result.clone(),
                            http_client_traces: None,
                        },
                    };
                    db_connection
                        .append_batch_respond_to_parent(
                            AppendEventsToExecution {
                                execution_id: ExecutionId::Derived(target_execution_id.clone()),
                                version: stub_finished_version.clone(),
                                batch: vec![finished_req],
                            },
                            AppendResponseToExecution {
                                parent_execution_id: parent_id,
                                created_at: called_at,
                                join_set_id,
                                child_execution_id: target_execution_id.clone(),
                                finished_version: stub_finished_version.clone(),
                                result: result.clone(),
                            },
                            called_at,
                        )
                        .await
                };
                debug!(%target_ffqn, %target_execution_id, "Executed append_batch_respond_to_parent: {write_attempt:?}");
                // The server might crash at this point, and restart processing.
                let persist_result = match write_attempt {
                    Ok(_) => Ok(()),
                    Err(err) => {
                        // TODO: check error == conflict
                        info!(%target_ffqn, %target_execution_id,
                            "append_batch_respond_to_parent was not successful, checking execution result - {err:?}"
                        );
                        // In case of conflict, select row (target_execution_id, version:1)
                        let found = db_connection
                            .get_execution_event(
                                &ExecutionId::Derived(target_execution_id.clone()),
                                &stub_finished_version,
                            )
                            .await?; // Not found at this point should not happen, unless the previous write failed. Will be retried.
                        match found.event {
                            ExecutionEventInner::Finished {
                                result: found_result,
                                ..
                            } if result == found_result => Ok(()),
                            ExecutionEventInner::Finished { .. } => {
                                info!(%target_ffqn, %target_execution_id, "Different value found in stubbed execution's finished event");
                                Err(())
                            }
                            other => {
                                info!(%target_ffqn, %target_execution_id,
                                    "Unexpected execution event at stubbed execution - {other:?}"
                                );
                                Err(())
                            }
                        }
                    }
                };

                // Second write tx: Append the HistoryEvent with target_result.
                let event = HistoryEvent::Stub {
                    target_execution_id: target_execution_id.clone(),
                    result,
                    persist_result,
                };
                let history_event = (event.clone(), db_connection.version.clone());
                let history_event_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                db_connection
                    .append_batch(
                        called_at,
                        vec![history_event_req],
                        db_connection.execution_id.clone(),
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;
                Ok(history_event)
            }
        }
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_to_db_blocking(
        &mut self,
        event_call: EventCallBlocking,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
        lock_expires_at: DateTime<Utc>,
    ) -> Result<Vec<(HistoryEvent, Version)>, DbErrorWrite> {
        trace!("append_to_db_blocking {}", db_connection.version);
        match event_call {
            EventCallBlocking::JoinNext(JoinNext {
                join_set_id,
                closing,
                wasm_backtrace,
            }) => {
                debug!(%join_set_id, "JoinNext(closing:{closing}): Flushing and appending JoinNext");
                let event =
                    if self.count_submissions(&join_set_id) > self.count_join_nexts(&join_set_id) {
                        HistoryEvent::JoinNext {
                            join_set_id,
                            run_expires_at: lock_expires_at,
                            requested_ffqn: None,
                            closing,
                        }
                    } else {
                        HistoryEvent::JoinNextTooMany {
                            join_set_id,
                            requested_ffqn: None,
                        }
                    };
                let history_events = vec![(event.clone(), db_connection.version.clone())];
                let join_next = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                db_connection
                    .append_blocking(
                        db_connection.execution_id.clone(),
                        join_next,
                        called_at,
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;
                Ok(history_events)
            }

            EventCallBlocking::JoinNextRequestingFfqn(JoinNextRequestingFfqn {
                join_set_id,
                requested_ffqn,
                wasm_backtrace,
            }) => {
                debug!(%join_set_id, "BlockingChildAwaitNext: Flushing and appending JoinNext");
                let event =
                    if self.count_submissions(&join_set_id) > self.count_join_nexts(&join_set_id) {
                        HistoryEvent::JoinNext {
                            join_set_id,
                            run_expires_at: lock_expires_at,
                            requested_ffqn: Some(requested_ffqn),
                            closing: false,
                        }
                    } else {
                        HistoryEvent::JoinNextTooMany {
                            join_set_id,
                            requested_ffqn: Some(requested_ffqn),
                        }
                    };
                let history_events = vec![(event.clone(), db_connection.version.clone())];
                let append_request = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                db_connection
                    .append_blocking(
                        db_connection.execution_id.clone(),
                        append_request,
                        called_at,
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;

                Ok(history_events)
            }

            EventCallBlocking::OneOffChildExecutionRequest(OneOffChildExecutionRequest {
                ffqn,
                fn_component_id,
                join_set_id,
                child_execution_id,
                params,
                wasm_backtrace,
            }) => {
                debug!(%child_execution_id, %join_set_id,
                    "OneOffChildExecutionRequest: Flushing and appending JoinSet,ChildExecutionRequest,JoinNext");
                let mut history_events = Vec::with_capacity(3);
                let event = HistoryEvent::JoinSetCreate {
                    join_set_id: join_set_id.clone(),
                };
                let mut version = db_connection.version.clone();
                history_events.push((event.clone(), version.clone()));
                let join_set = AppendRequest {
                    event: ExecutionEventInner::HistoryEvent { event },
                    created_at: called_at,
                };
                let event = HistoryEvent::JoinSetRequest {
                    join_set_id: join_set_id.clone(),
                    request: JoinSetRequest::ChildExecutionRequest {
                        child_execution_id: child_execution_id.clone(),
                        target_ffqn: ffqn.clone(),
                        params: params.clone(),
                    },
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let child_exec_req = AppendRequest {
                    event: ExecutionEventInner::HistoryEvent { event },
                    created_at: called_at,
                };
                let event = HistoryEvent::JoinNext {
                    join_set_id: join_set_id.clone(),
                    run_expires_at: lock_expires_at,
                    requested_ffqn: Some(ffqn.clone()),
                    closing: false,
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let join_next = AppendRequest {
                    event: ExecutionEventInner::HistoryEvent { event },
                    created_at: called_at,
                };

                let child = CreateRequest {
                    created_at: called_at,
                    execution_id: ExecutionId::Derived(child_execution_id),
                    ffqn,
                    params,
                    parent: Some((db_connection.execution_id.clone(), join_set_id)),
                    metadata: ExecutionMetadata::from_parent_span(&self.worker_span),
                    scheduled_at: called_at,
                    component_id: fn_component_id,
                    scheduled_by: None,
                };
                db_connection
                    .append_batch_create_new_execution(
                        called_at,
                        vec![join_set, child_exec_req, join_next],
                        db_connection.execution_id.clone(),
                        vec![child],
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;

                Ok(history_events)
            }

            EventCallBlocking::OneOffDelayRequest(OneOffDelayRequest {
                join_set_id,
                delay_id,
                schedule_at,
                expires_at_if_new,
                wasm_backtrace,
            }) => {
                debug!(%delay_id, %join_set_id, "BlockingDelayRequest: Flushing and appending JoinSet,DelayRequest,JoinNext");
                let mut history_events = Vec::with_capacity(3);
                let event = HistoryEvent::JoinSetCreate {
                    join_set_id: join_set_id.clone(),
                };
                let mut version = db_connection.version.clone();
                history_events.push((event.clone(), version.clone()));
                let join_set = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let event = HistoryEvent::JoinSetRequest {
                    join_set_id: join_set_id.clone(),
                    request: JoinSetRequest::DelayRequest {
                        delay_id,
                        expires_at: expires_at_if_new,
                        schedule_at,
                    },
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let delay_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };
                let event = HistoryEvent::JoinNext {
                    join_set_id,
                    run_expires_at: lock_expires_at,
                    closing: false,
                    requested_ffqn: None,
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let join_next = AppendRequest {
                    created_at: called_at,
                    event: ExecutionEventInner::HistoryEvent { event },
                };

                db_connection
                    .append_batch(
                        called_at,
                        vec![join_set, delay_req, join_next],
                        db_connection.execution_id.clone(),
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;

                Ok(history_events)
            }
        }
    }

    #[expect(clippy::result_large_err)]
    pub(crate) fn get_processed_response(
        &self,
        child_execution_id: &ExecutionIdDerived,
        specified_ffqn: &FunctionFqn,
    ) -> Result<WastVal, GetExtensionError> {
        let found_ffqn = self
            .index_child_exe_to_ffqn
            .get(child_execution_id)
            .ok_or(GetExtensionError::NotFoundInProcessedResponses)?; // Not even found in unprocessed requests.
        let response_idx = self
            .index_child_exe_to_processed_response_idx
            .get(child_execution_id)
            .ok_or(GetExtensionError::NotFoundInProcessedResponses)?;

        if specified_ffqn != found_ffqn {
            return Err(GetExtensionError::FunctionMismatch(
                types_execution::FunctionMismatch {
                    specified_function: types_execution::Function::from(specified_ffqn),
                    actual_function: Some(types_execution::Function::from(found_ffqn)),
                    actual_id: types_execution::ResponseId::ExecutionId(
                        types_execution::ExecutionId::from(child_execution_id),
                    ),
                },
            ));
        }
        match &self
            .responses
            .get(*response_idx)
            .as_ref()
            .expect("`index_child_exe_to_processed_response_idx` must point to a response")
            .0
            .event
        {
            JoinSetResponse::ChildExecutionFinished {
                result,
                child_execution_id,
                finished_version: _,
            } => {
                let response_ffqn = self
                    .index_child_exe_to_ffqn
                    .get(child_execution_id)
                    .expect("got response so the request must have been processed");
                Ok(result
                    .clone()
                    .into_wast_val( || self.fn_registry.get_ret_type(response_ffqn)
                        .expect("response_ffqn can only be exported and no-ext, thus must be returned by get_ret_type"))

                )
            }
            JoinSetResponse::DelayFinished { .. } => unreachable!(
                "`index_child_exe_to_processed_response_idx` must point to a ChildExecutionFinished"
            ),
        }
    }

    pub(crate) fn next_join_set_name_generated(&self) -> String {
        self.next_join_set_name_index(JoinSetKind::Generated)
    }

    fn next_join_set_name_index(&self, kind: JoinSetKind) -> String {
        assert!(kind != JoinSetKind::Named);
        (self.join_set_count(kind) + 1).to_string()
    }

    fn next_join_set_one_off_named(
        &self,
        suffix: &str,
    ) -> Result<JoinSetId, InvalidNameError<JoinSetId>> {
        let index = self.next_join_set_name_index(JoinSetKind::OneOff);
        JoinSetId::new(
            JoinSetKind::OneOff,
            StrVariant::from(format!("{index}-{suffix}")),
        )
    }

    fn count_submissions(&self, join_set_id: &JoinSetId) -> usize {
        // Processing status does matter in order to make replays make the same decisions.
        self.event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                *processing_status == Processed
                    && match event {
                        HistoryEvent::JoinSetRequest {
                            join_set_id: found_join_set_id,
                            ..
                        } => found_join_set_id == join_set_id,
                        HistoryEvent::Persist { .. }
                        | HistoryEvent::JoinSetCreate { .. }
                        | HistoryEvent::JoinNext { .. }
                        | HistoryEvent::JoinNextTooMany { .. }
                        | HistoryEvent::Schedule { .. }
                        | HistoryEvent::Stub { .. } => false,
                    }
            })
            .count()
    }

    fn count_join_nexts(&self, join_set_id: &JoinSetId) -> usize {
        // Processing status does matter in order to make replays make the same decisions.
        self.event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                *processing_status == Processed
                    && match event {
                        HistoryEvent::JoinNext {
                            join_set_id: found_join_set_id,
                            ..
                        } => found_join_set_id == join_set_id,
                        HistoryEvent::Persist { .. }
                        | HistoryEvent::JoinSetCreate { .. }
                        | HistoryEvent::JoinSetRequest { .. }
                        | HistoryEvent::JoinNextTooMany { .. }
                        | HistoryEvent::Schedule { .. }
                        | HistoryEvent::Stub { .. } => false,
                    }
            })
            .count()
    }

    pub(crate) fn next_delay_id(
        &self,
        join_set_id: &JoinSetId,
        execution_id: &ExecutionId,
    ) -> DelayId {
        // TODO: Optimize
        let idx = self
            .event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                // Ignore unprocessed events, which happens only on replay where we need to produce "old" ids.
                // Applying the `EventCall::SubmitDelay` will immediately mark this as processed.
                *processing_status == Processed &&
                matches!(
                    event,
                    HistoryEvent::JoinSetRequest {join_set_id:found_join_set_id, request:JoinSetRequest::DelayRequest { .. } }
                    if join_set_id == found_join_set_id
                )
            })
            .count();
        DelayId::new_with_index(
            execution_id,
            join_set_id,
            u64::try_from(idx).expect("too many delays in a join set"),
        )
    }
}

#[derive(Debug, PartialEq, Eq)]
enum AwaitNextExtensionError {
    FunctionMismatch {
        specified_function: FunctionFqn,
        actual_function: Option<FunctionFqn>, // None on delay
        actual_id: ResponseId,
    },
    AllProcessed,
}
impl AwaitNextExtensionError {
    fn as_wast_val_result(&self) -> WastVal {
        WastVal::Result(Err(Some(Box::new(self.as_wast_val_internal()))))
    }

    fn as_wast_val_internal(&self) -> WastVal {
        match self {
            AwaitNextExtensionError::FunctionMismatch {
                specified_function: specified,
                actual_function: actual,
                actual_id,
            } => {
                let (actual_id_field_name, actual_str_value) = match actual_id {
                    ResponseId::ChildExecutionId(id) => ("execution-id", id.to_string()),
                    ResponseId::DelayId(id) => ("delay-id", id.to_string()),
                };
                WastVal::Variant(
                    "function-mismatch".to_string(),
                    Some(Box::new(WastVal::Record(indexmap! {
                        "specified-function".to_string() => ffqn_into_wast_val(specified),
                        "actual-function".to_string() =>  WastVal::Option(
                            actual.as_ref().map(|actual| Box::from(ffqn_into_wast_val(actual)))),
                        "actual-id".to_string() =>
                            WastVal::Variant(actual_id_field_name.to_string(),
                            Some(Box::new(
                            WastVal::Record(indexmap!{
                                "id".to_string() => WastVal::String(actual_str_value)
                            }))
                        ))
                    }))),
                )
            }
            AwaitNextExtensionError::AllProcessed => {
                WastVal::Variant("all-processed".to_string(), None)
            }
        }
    }
}

enum JoinSetResponseEnriched<'a> {
    DelayFinished {
        delay_id: &'a DelayId,
        expires_at: DateTime<Utc>,
        result: Result<(), ()>,
    },
    ChildExecutionFinished(ChildExecutionFinished<'a>),
}

struct ChildExecutionFinished<'a> {
    child_execution_id: &'a ExecutionIdDerived,
    result: &'a SupportedFunctionReturnValue,
    response_ffqn: &'a FunctionFqn,
}

#[derive(Debug)]
enum JoinNextVariant {
    Child {
        join_set_id: JoinSetId,
        kind: JoinNextChildKind,
        requested_ffqn: FunctionFqn, // For storing index_child_exe_to_ffqn
    },
    Delay(JoinSetId),
    JoinNext {
        join_set_id: JoinSetId,
        closing: bool,
    },
}
impl JoinNextVariant {
    fn join_set_id(&self) -> &JoinSetId {
        match self {
            JoinNextVariant::Child { join_set_id, .. }
            | JoinNextVariant::Delay(join_set_id)
            | JoinNextVariant::JoinNext {
                join_set_id,
                closing: _,
            } => join_set_id,
        }
    }
    fn as_key(&self) -> DeterministicKey {
        match self {
            JoinNextVariant::Child {
                join_set_id,
                kind,
                requested_ffqn,
            } => DeterministicKey::JoinNextChild {
                join_set_id: join_set_id.clone(),
                kind: *kind,
                requested_ffqn: requested_ffqn.clone(),
            },
            JoinNextVariant::Delay(join_set_id) => DeterministicKey::JoinNextDelay {
                join_set_id: join_set_id.clone(),
            },
            JoinNextVariant::JoinNext {
                join_set_id,
                closing,
            } => DeterministicKey::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: *closing,
            },
        }
    }
}

#[derive(derive_more::Debug, Clone, IntoStaticStr)]
pub(crate) enum EventCall {
    Blocking(EventCallBlocking),
    NonBlocking(EventCallNonBlocking),
}

#[derive(derive_more::Debug, Clone, IntoStaticStr)]
pub(crate) enum EventCallBlocking {
    /// foo-await-next: func(join-set: borrow<join-set>) -> result<tuple<execution-id, ?>, await-next-extension-error>;
    JoinNextRequestingFfqn(JoinNextRequestingFfqn),
    JoinNext(JoinNext),
    OneOffChildExecutionRequest(OneOffChildExecutionRequest), // blocking call
    OneOffDelayRequest(OneOffDelayRequest),                   // blocking sleep
}

#[derive(derive_more::Debug, Clone, IntoStaticStr)]
pub(crate) enum EventCallNonBlocking {
    JoinSetCreate(JoinSetCreate),
    SubmitChildExecution(SubmitChildExecution),
    SubmitDelay(SubmitDelay),
    Schedule(Schedule),
    Stub(Stub),
    Persist(Persist),
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct JoinSetCreate {
    pub(crate) join_set_id: JoinSetId,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl JoinSetCreate {
    // The caller must perform conflict check
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<JoinSetId, ApplyError> {
        assert!(self.join_set_id.kind != JoinSetKind::OneOff);
        let join_set_id = self.join_set_id.clone();
        let value = event_history
            .apply_inner(
                EventCall::NonBlocking(EventCallNonBlocking::JoinSetCreate(self)),
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value,
            ChildReturnValue::JoinSetCreate(join_set_id) => join_set_id);
        assert_eq!(join_set_id, value);
        let prev_val = event_history
            .index_join_set_to_unawaited_requests
            .insert(join_set_id, IndexMap::default());
        assert!(
            prev_val.is_none(),
            "conflict check must have been performed by the caller"
        );
        Ok(value)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct SubmitChildExecution {
    pub(crate) target_ffqn: FunctionFqn,
    pub(crate) fn_component_id: ComponentId,
    pub(crate) join_set_id: JoinSetId,
    pub(crate) child_execution_id: ExecutionIdDerived,
    #[debug(skip)]
    pub(crate) params: Params,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl SubmitChildExecution {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Val /* ExecutionId */, WorkflowFunctionError> {
        assert!(
            self.join_set_id.kind != JoinSetKind::OneOff,
            "one-off join set cannot be constructed outside of OneOff*Request"
        );
        let join_set_id = self.join_set_id.clone();
        let child_execution_id = self.child_execution_id.clone();
        let component_type = self.fn_component_id.component_type;
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::SubmitChildExecution(self)),
                db_connection,
                called_at,
            )
            .await?;
        // TODO: return void
        let value =
            assert_matches!(value, ChildReturnValue::WastVal(wast_val) => wast_val.as_val());

        event_history
            .index_join_set_to_unawaited_requests
            .get_mut(&join_set_id)
            .ok_or_else(|| {
                WorkflowFunctionError::ConstraintViolation(
                    format!("not found in open join sets: `{join_set_id}`").into(),
                )
            })?
            .insert(
                ResponseId::ChildExecutionId(child_execution_id),
                component_type,
            );

        Ok(value)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct SubmitDelay {
    pub(crate) join_set_id: JoinSetId,
    pub(crate) delay_id: DelayId,
    pub(crate) schedule_at: HistoryEventScheduleAt, // Intention that must be compared when checking determinism
    pub(crate) expires_at_if_new: DateTime<Utc>, // Actual time based on first execution. Should be disregarded on replay.
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl SubmitDelay {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<DelayId, WorkflowFunctionError> {
        assert!(
            self.join_set_id.kind != JoinSetKind::OneOff,
            "one-off join set cannot be constructed outside of OneOff*Request"
        );
        let join_set_id = self.join_set_id.clone();
        let delay_id = self.delay_id.clone();
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::SubmitDelay(self)),
                db_connection,
                called_at,
            )
            .await?;
        assert_matches!(value, ChildReturnValue::SubmitDelay);
        event_history
            .index_join_set_to_unawaited_requests
            .get_mut(&join_set_id)
            .ok_or_else(|| {
                WorkflowFunctionError::ConstraintViolation(
                    format!("not found in open join sets: `{join_set_id}`").into(),
                )
            })?
            .insert(
                ResponseId::DelayId(delay_id.clone()),
                INVALID_CHILD_TYPE_FOR_DELAYS,
            );
        Ok(delay_id)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct Schedule {
    #[expect(clippy::struct_field_names)]
    pub(crate) schedule_at: HistoryEventScheduleAt, // Intention that must be compared when checking determinism
    pub(crate) scheduled_at_if_new: DateTime<Utc>, // Actual time based on first execution. Should be disregarded on replay.
    pub(crate) execution_id: ExecutionId,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) fn_component_id: ComponentId,
    #[debug(skip)]
    pub(crate) params: Params,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl Schedule {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val /* ExecutionId */, WorkflowFunctionError> {
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::Schedule(self)),
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value,
            ChildReturnValue::WastVal(wast_val) => wast_val.as_val());
        Ok(value)
    }
}

// -stub: func(execution-id: execution-id, return-value: result<?>) -> result<_, stub-error>
#[derive(derive_more::Debug, Clone)]
pub(crate) struct Stub {
    pub(crate) target_ffqn: FunctionFqn,
    pub(crate) target_execution_id: ExecutionIdDerived,
    pub(crate) parent_id: ExecutionId,
    pub(crate) join_set_id: JoinSetId,
    #[debug(skip)]
    pub(crate) result: SupportedFunctionReturnValue, // stubbed return value
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl Stub {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::Stub(self)),
                db_connection,
                called_at,
            )
            .await?;

        let value =
            assert_matches!(value, ChildReturnValue::WastVal(wast_val) => wast_val.as_val());
        Ok(value)
    }
}

// -await-next: func(join-set-id: borrow<join-set-id>) -> result<(execution_id,?), await-next-extension-error>
#[derive(derive_more::Debug, Clone)]
pub(crate) struct JoinNextRequestingFfqn {
    pub(crate) join_set_id: JoinSetId,
    pub(crate) requested_ffqn: FunctionFqn,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl JoinNextRequestingFfqn {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<
        wasmtime::component::Val, /* result<?, await-next-extension-error> */
        WorkflowFunctionError,
    > {
        assert!(
            self.join_set_id.kind != JoinSetKind::OneOff,
            "one-off join set cannot be constructed outside of OneOff*Request"
        );
        let join_set_id = self.join_set_id.clone();
        let value = event_history
            .apply(
                EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(self)),
                db_connection,
                called_at,
            )
            .await?;

        let value =
            assert_matches!(value, ChildReturnValue::JoinNextRequestingFfqn(result) => result);
        let value = match value {
            Ok((child_execution_id, wast_val_result)) => {
                let wast_val_res = WastVal::Result(Ok(Some(Box::new(WastVal::Tuple(vec![
                    execution_id_derived_into_wast_val(&child_execution_id),
                    wast_val_result,
                ])))));
                let was_present = event_history
                    .index_join_set_to_unawaited_requests
                    .get_mut(&join_set_id)
                    .ok_or_else(|| {
                        WorkflowFunctionError::ConstraintViolation(
                            format!("not found in open join sets: `{join_set_id}`").into(),
                        )
                    })?
                    .shift_remove(&ResponseId::ChildExecutionId(child_execution_id));
                assert!(was_present.is_some());
                wast_val_res
            }
            Err(await_ext_err) => {
                if let AwaitNextExtensionError::FunctionMismatch { actual_id, .. } = &await_ext_err
                {
                    let was_present = event_history
                        .index_join_set_to_unawaited_requests
                        .get_mut(&join_set_id)
                        .ok_or_else(|| {
                            WorkflowFunctionError::ConstraintViolation(
                                format!("not found in open join sets: `{join_set_id}`").into(),
                            )
                        })?
                        .shift_remove(actual_id);
                    assert!(was_present.is_some());
                }
                await_ext_err.as_wast_val_result()
            }
        }
        .as_val();
        Ok(value)
    }
}

// join-next: func(join-set-id: borrow<join-set-id>) -> result<response-id, join-next-error>
#[derive(derive_more::Debug, Clone)]
pub(crate) struct JoinNext {
    pub(crate) join_set_id: JoinSetId,
    pub(crate) closing: bool,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl JoinNext {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<
        Result<(types_execution::ResponseId, Result<(), ()>), types_join_set::JoinNextError>,
        WorkflowFunctionError,
    > {
        assert!(
            self.join_set_id.kind != JoinSetKind::OneOff,
            "one-off join set cannot be constructed outside of OneOff*Request"
        );
        let join_set_id = self.join_set_id.clone();
        let value = event_history
            .apply_inner(
                EventCall::Blocking(EventCallBlocking::JoinNext(self)),
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value,ChildReturnValue::JoinNext(value) => value);
        if let Ok((response_id, _)) = &value {
            let was_present = event_history
                .index_join_set_to_unawaited_requests
                .get_mut(&join_set_id)
                .ok_or_else(|| {
                    WorkflowFunctionError::ConstraintViolation(
                        format!("not found in open join sets: `{join_set_id}`").into(),
                    )
                })?
                .shift_remove(response_id);
            assert!(was_present.is_some());
        }
        let value = value
            .map(|(response_id, result)| (types_execution::ResponseId::from(response_id), result));
        Ok(value)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct OneOffChildExecutionRequest {
    ffqn: FunctionFqn,
    fn_component_id: ComponentId,
    join_set_id: JoinSetId,
    child_execution_id: ExecutionIdDerived,
    #[debug(skip)]
    params: Params,
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl OneOffChildExecutionRequest {
    pub(crate) async fn apply(
        ffqn: FunctionFqn,
        fn_component_id: ComponentId,
        params: Params,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Val, WorkflowFunctionError> {
        let join_set_id = event_history
            .next_join_set_one_off_named(&ffqn.function_name)
            .expect("no illegal chars are allowed in fn name by WIT, only alphanumeric and dash");
        let child_execution_id = db_connection.execution_id.next_level(&join_set_id);
        let event = EventCall::Blocking(EventCallBlocking::OneOffChildExecutionRequest(
            OneOffChildExecutionRequest {
                ffqn,
                fn_component_id,
                join_set_id,
                child_execution_id,
                params,
                wasm_backtrace,
            },
        ));

        let value = event_history.apply(event, db_connection, called_at).await?;
        let value = assert_matches!(value,
            ChildReturnValue::WastVal(wast_val) => wast_val.as_val());
        Ok(value)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn apply_invoke(
        ffqn: FunctionFqn,
        fn_component_id: ComponentId,
        label: &str,
        params: Params,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Val, WorkflowFunctionError> {
        let join_set_id = match event_history.next_join_set_one_off_named(label) {
            Ok(ok) => ok,
            Err(err) => {
                // invoke-extension-error
                return Ok(Val::Result(Err(Some(Box::new(Val::Variant(
                    "invalid-name".to_string(),
                    Some(Box::new(Val::String(err.to_string()))),
                ))))));
            }
        };
        let child_execution_id = db_connection.execution_id.next_level(&join_set_id);
        let event = EventCall::Blocking(EventCallBlocking::OneOffChildExecutionRequest(
            OneOffChildExecutionRequest {
                ffqn,
                fn_component_id,
                join_set_id,
                child_execution_id: child_execution_id.clone(),
                params,
                wasm_backtrace,
            },
        ));
        let value = event_history
            .apply_inner(event, db_connection, called_at)
            .await?;
        let value = assert_matches!(value, ChildReturnValue::WastVal(wast_val) => {
            // wrap with an Ok and a tuple in order to return result<tuple<execution-id, T>, invoke-extension-error>
            Val::Result(Ok(Some(
                Box::new(
                Val::Tuple(vec![
                    execution_id_derived_into_wast_val(&child_execution_id).as_val(),
                    wast_val.as_val()
            ])))))
        });
        Ok(value)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct OneOffDelayRequest {
    join_set_id: JoinSetId,
    delay_id: DelayId,
    schedule_at: HistoryEventScheduleAt, // Intention that must be compared when checking determinism
    expires_at_if_new: DateTime<Utc>, // Actual time based on first execution. Should be disregarded on replay.
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl OneOffDelayRequest {
    pub(crate) async fn apply(
        schedule_at: HistoryEventScheduleAt,
        expires_at_if_new: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Result<DateTime<Utc>, ()>, WorkflowFunctionError> {
        let join_set_id = event_history
            .next_join_set_one_off_named("sleep")
            .expect("no illegal chars in sleep");
        let delay_id = DelayId::new(&db_connection.execution_id, &join_set_id);

        let ChildReturnValue::OneOffDelay {
            scheduled_at,
            result,
        } = event_history
            .apply(
                EventCall::Blocking(EventCallBlocking::OneOffDelayRequest(OneOffDelayRequest {
                    join_set_id,
                    delay_id,
                    schedule_at,
                    expires_at_if_new,
                    wasm_backtrace,
                })),
                db_connection,
                called_at,
            )
            .await?
        else {
            unreachable!()
        };
        Ok(result.map(|()| scheduled_at))
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct Persist {
    #[debug(skip)]
    pub(crate) value: Vec<u8>,
    pub(crate) kind: PersistKind,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl Persist {
    pub(crate) async fn apply_string(
        value: String,
        min_length: u64,
        max_length_exclusive: u64,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<String, WorkflowFunctionError> {
        let value = Vec::from_iter(value.bytes());
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::Persist(Persist {
                    value,
                    kind: PersistKind::RandomString {
                        min_length,
                        max_length_exclusive,
                    },
                    wasm_backtrace,
                })),
                db_connection,
                called_at,
            )
            .await?;
        let value =
            assert_matches!(value, ChildReturnValue::WastVal(WastVal::String(value)) => value);
        Ok(value)
    }

    pub(crate) async fn apply_u64(
        value: u64,
        min: u64,
        max_inclusive: u64,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        db_connection: &mut CachingDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<u64, WorkflowFunctionError> {
        let value = Vec::from(storage::from_u64_to_bytes(value));
        let value = event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::Persist(Persist {
                    value,
                    kind: PersistKind::RandomU64 { min, max_inclusive },
                    wasm_backtrace,
                })),
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value, ChildReturnValue::WastVal(WastVal::U64(value)) => value);
        Ok(value)
    }
}

impl Display for EventCall {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EventCall::Blocking(inner) => write!(f, "{}", <&str>::from(inner)),
            EventCall::NonBlocking(inner) => write!(f, "{}", <&str>::from(inner)),
        }
    }
}

impl EventCallBlocking {
    fn join_next_variant(&self) -> JoinNextVariant {
        match &self {
            // Blocking calls can be polled for JoinSetResponse
            EventCallBlocking::OneOffChildExecutionRequest(OneOffChildExecutionRequest {
                join_set_id,
                ffqn,
                fn_component_id: _,
                child_execution_id: _,
                params: _,
                wasm_backtrace: _,
            }) => JoinNextVariant::Child {
                join_set_id: join_set_id.clone(),
                kind: JoinNextChildKind::DirectCall,
                requested_ffqn: ffqn.clone(),
            },
            EventCallBlocking::JoinNextRequestingFfqn(JoinNextRequestingFfqn {
                join_set_id,
                requested_ffqn,
                wasm_backtrace: _,
            }) => JoinNextVariant::Child {
                join_set_id: join_set_id.clone(),
                kind: JoinNextChildKind::AwaitNext,
                requested_ffqn: requested_ffqn.clone(),
            },
            EventCallBlocking::OneOffDelayRequest(OneOffDelayRequest {
                join_set_id,
                delay_id: _,
                schedule_at: _,
                expires_at_if_new: _,
                wasm_backtrace: _,
            }) => JoinNextVariant::Delay(join_set_id.clone()),
            EventCallBlocking::JoinNext(JoinNext {
                join_set_id,
                closing,
                wasm_backtrace: _,
            }) => JoinNextVariant::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: *closing,
            },
        }
    }
}

/// Stores important properties of events as they are executed or replayed.
/// Those properties are checked for equality with `HistoryEvent` before it is marked as `Processed`.
/// One `EventCall` can be represented as multiple `DeterministicKey`-s.
/// One `DeterministicKey` corresponds to one `HistoryEvent`.
#[derive(derive_more::Debug, Clone, derive_more::Display)]
enum DeterministicKey {
    #[display("Persist({kind})")]
    Persist {
        #[debug(skip)]
        value: Vec<u8>,
        kind: PersistKind,
    },

    #[display("CreateJoinSet({join_set_id})")]
    CreateJoinSet { join_set_id: JoinSetId },

    #[display("ChildExecutionRequest({child_execution_id}, {target_ffqn}, params: {params})")]
    // join_set_id is part of child_execution_id
    ChildExecutionRequest {
        join_set_id: JoinSetId,
        child_execution_id: ExecutionIdDerived,
        target_ffqn: FunctionFqn,
        params: Params,
    },

    #[display("DelayRequest({delay_id}, {schedule_at})")] // join_set_id is part of delay_id
    DelayRequest {
        join_set_id: JoinSetId,
        delay_id: DelayId,
        schedule_at: HistoryEventScheduleAt,
    },

    #[display("JoinNextChild({join_set_id}, {kind}, {requested_ffqn})")]
    JoinNextChild {
        // TODO: Rename to JoinNextRequestingFfqn
        join_set_id: JoinSetId,
        kind: JoinNextChildKind,
        requested_ffqn: FunctionFqn,
    },

    // Can only be part of blocking (one-off) sleep
    #[display("JoinNextDelay({join_set_id})")]
    JoinNextDelay { join_set_id: JoinSetId },

    #[display("JoinNext({join_set_id}{})", if *closing {" closing"} else {""} )]
    JoinNext {
        join_set_id: JoinSetId,
        closing: bool,
    },

    #[display("Schedule({target_execution_id}, {schedule_at})")]
    Schedule {
        target_execution_id: ExecutionId,
        schedule_at: HistoryEventScheduleAt,
    },

    #[display("Stub({target_execution_id})")]
    Stub {
        target_execution_id: ExecutionIdDerived,
        return_value: SupportedFunctionReturnValue,
    },
}

#[derive(Debug, Clone, Copy, derive_more::Display)]
enum JoinNextChildKind {
    AwaitNext,
    DirectCall,
}

impl EventCall {
    fn as_keys(&self) -> Vec<DeterministicKey> {
        match self {
            EventCall::Blocking(inner) => inner.as_keys(),
            EventCall::NonBlocking(inner) => vec![inner.as_key()],
        }
    }
}

impl EventCallBlocking {
    fn as_keys(&self) -> Vec<DeterministicKey> {
        match self {
            EventCallBlocking::JoinNextRequestingFfqn(JoinNextRequestingFfqn {
                join_set_id,
                requested_ffqn,
                ..
            }) => {
                vec![DeterministicKey::JoinNextChild {
                    join_set_id: join_set_id.clone(),
                    kind: JoinNextChildKind::AwaitNext,
                    requested_ffqn: requested_ffqn.clone(),
                }]
            }
            EventCallBlocking::OneOffChildExecutionRequest(OneOffChildExecutionRequest {
                join_set_id,
                child_execution_id,
                ffqn,
                params,
                fn_component_id: _,
                wasm_backtrace: _,
            }) => vec![
                DeterministicKey::CreateJoinSet {
                    join_set_id: join_set_id.clone(),
                },
                DeterministicKey::ChildExecutionRequest {
                    join_set_id: join_set_id.clone(),
                    child_execution_id: child_execution_id.clone(),
                    target_ffqn: ffqn.clone(),
                    params: params.clone(),
                },
                DeterministicKey::JoinNextChild {
                    join_set_id: join_set_id.clone(),
                    kind: JoinNextChildKind::DirectCall,
                    requested_ffqn: ffqn.clone(),
                },
            ],
            EventCallBlocking::OneOffDelayRequest(OneOffDelayRequest {
                join_set_id,
                delay_id,
                schedule_at,
                ..
            }) => vec![
                DeterministicKey::CreateJoinSet {
                    join_set_id: join_set_id.clone(),
                },
                DeterministicKey::DelayRequest {
                    join_set_id: join_set_id.clone(),
                    delay_id: delay_id.clone(),
                    schedule_at: *schedule_at,
                },
                DeterministicKey::JoinNextDelay {
                    join_set_id: join_set_id.clone(),
                },
            ],
            EventCallBlocking::JoinNext(JoinNext {
                join_set_id,
                closing,
                wasm_backtrace: _,
            }) => vec![DeterministicKey::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: *closing,
            }],
        }
    }
}

impl EventCallNonBlocking {
    fn as_key(&self) -> DeterministicKey {
        match self {
            EventCallNonBlocking::JoinSetCreate(JoinSetCreate { join_set_id, .. }) => {
                DeterministicKey::CreateJoinSet {
                    join_set_id: join_set_id.clone(),
                }
            }
            EventCallNonBlocking::Persist(Persist { value, kind, .. }) => {
                DeterministicKey::Persist {
                    value: value.clone(),
                    kind: *kind,
                }
            }
            EventCallNonBlocking::SubmitChildExecution(SubmitChildExecution {
                join_set_id,
                child_execution_id,
                target_ffqn,
                params,
                fn_component_id: _,
                wasm_backtrace: _,
            }) => DeterministicKey::ChildExecutionRequest {
                join_set_id: join_set_id.clone(),
                child_execution_id: child_execution_id.clone(),
                target_ffqn: target_ffqn.clone(),
                params: params.clone(),
            },
            EventCallNonBlocking::SubmitDelay(SubmitDelay {
                delay_id,
                join_set_id,
                schedule_at: timeout,
                ..
            }) => DeterministicKey::DelayRequest {
                join_set_id: join_set_id.clone(),
                delay_id: delay_id.clone(),
                schedule_at: *timeout,
            },
            EventCallNonBlocking::Schedule(Schedule {
                execution_id,
                schedule_at,
                ..
            }) => DeterministicKey::Schedule {
                target_execution_id: execution_id.clone(),
                schedule_at: *schedule_at,
            },
            EventCallNonBlocking::Stub(Stub {
                target_execution_id,
                result: return_value,
                ..
            }) => DeterministicKey::Stub {
                target_execution_id: target_execution_id.clone(),
                return_value: return_value.clone(),
            },
        }
    }
}

mod response_id {
    use concepts::{
        ComponentType,
        prefixed_ulid::{DelayId, ExecutionIdDerived},
    };
    use std::hash::Hash;

    use crate::workflow::host_exports::v4_0_0::{
        DelayId_4_0_0, ExecutionId_4_0_0, ResponseId_4_0_0,
    };

    pub(crate) const INVALID_CHILD_TYPE_FOR_DELAYS: ComponentType = ComponentType::WebhookEndpoint;

    #[derive(Debug, PartialEq, Eq, Hash)]
    pub(crate) enum ResponseId {
        ChildExecutionId(ExecutionIdDerived),
        DelayId(DelayId),
    }

    impl From<ResponseId> for ResponseId_4_0_0 {
        fn from(value: crate::workflow::event_history::ResponseId) -> Self {
            use crate::workflow::event_history::ResponseId;
            match value {
                ResponseId::ChildExecutionId(child_execution_id) => {
                    ResponseId_4_0_0::ExecutionId(ExecutionId_4_0_0::from(&child_execution_id))
                }
                ResponseId::DelayId(delay_id) => {
                    ResponseId_4_0_0::DelayId(DelayId_4_0_0::from(&delay_id))
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::super::event_history::{
        EventCall, EventCallBlocking, EventCallNonBlocking, EventHistory,
    };
    use super::super::workflow_worker::JoinNextBlockingStrategy;
    use super::SubmitChildExecution;
    use crate::activity::cancel_registry::CancelRegistry;
    use crate::testing_fn_registry::TestingFnRegistry;
    use crate::workflow::caching_db_connection::{CachingBuffer, CachingDbConnection};
    use crate::workflow::deadline_tracker::DeadlineTrackerFactory;
    use crate::workflow::deadline_tracker::deadline_tracker_factory_test;
    use crate::workflow::event_history::response_id::ResponseId;
    use crate::workflow::event_history::{
        ApplyError, AwaitNextExtensionError, ChildReturnValue, JoinNextRequestingFfqn,
        JoinSetCreate, Schedule, Stub,
    };
    use assert_matches::assert_matches;
    use chrono::{DateTime, Utc};
    use concepts::prefixed_ulid::{ExecutionIdDerived, ExecutorId, RunId};
    use concepts::storage::{CreateRequest, HistoryEventScheduleAt, Locked};
    use concepts::storage::{
        DbConnection, DbPoolCloseable, JoinSetResponse, JoinSetResponseEvent, Version,
    };
    use concepts::time::ClockFn;
    use concepts::{
        ComponentId, ComponentRetryConfig, ExecutionId, FunctionFqn, FunctionRegistry, Params,
        SUPPORTED_RETURN_VALUE_OK_EMPTY, SupportedFunctionReturnValue,
    };
    use concepts::{JoinSetId, StrVariant};
    use db_tests::Database;
    use rstest::rstest;
    use std::sync::Arc;
    use std::time::Duration;
    use test_utils::sim_clock::SimClock;
    use tracing::{info, info_span};
    use val_json::type_wrapper::TypeWrapper;
    use val_json::wast_val::{WastVal, WastValWithType};

    pub const MOCK_FFQN: FunctionFqn = FunctionFqn::new_static("namespace:pkg/ifc", "fn1");
    pub const MOCK_FFQN_2: FunctionFqn = FunctionFqn::new_static("namespace:pkg/ifc", "fn2");

    #[rstest]
    #[tokio::test]
    async fn regular_join_next_child(
        #[values(JoinNextBlockingStrategy::Interrupt, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 0}, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 100})]
        second_run_strategy: JoinNextBlockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();

        // Create an execution.
        let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;

        let fn_registry = TestingFnRegistry::new_from_components(vec![]);

        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id.clone(),
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            JoinNextBlockingStrategy::Interrupt, // first run needs to interrupt
            fn_registry.clone(),
        )
        .await;

        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let child_execution_id = execution_id.next_level(&join_set_id);

        assert_matches!(
            apply_create_join_set_start_async_await_next(
                &mut caching_db_connection,
                MOCK_FFQN,
                child_execution_id.clone(),
                &mut event_history,
                join_set_id.clone(),
                sim_clock.now()
            )
            .await
            .unwrap_err(),
            ApplyError::InterruptDbUpdated,
            "should have ended with an interrupt"
        );
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: child_execution_id.clone(),
                        finished_version: Version(0), // does not matter
                        result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");
        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id,
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            second_run_strategy,
            fn_registry,
        )
        .await;
        apply_create_join_set_start_async_await_next(
            &mut caching_db_connection,
            MOCK_FFQN,
            child_execution_id,
            &mut event_history,
            join_set_id,
            sim_clock.now(),
        )
        .await
        .expect("response was appended, should finish successfuly");

        event_history
            .finalize(&mut caching_db_connection, sim_clock.now())
            .await
            .unwrap();

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

    #[rstest]
    #[tokio::test]
    async fn start_async_respond_then_join_next(
        #[values(JoinNextBlockingStrategy::Interrupt, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 0}, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 10})]
        join_next_blocking_strategy: JoinNextBlockingStrategy,
    ) {
        const CHILD_RESP: SupportedFunctionReturnValue = SupportedFunctionReturnValue::Ok {
            ok: Some(WastValWithType {
                r#type: TypeWrapper::U8,
                value: WastVal::U8(1),
            }),
        };
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();
        // Create an execution.
        let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;
        let fn_registry = TestingFnRegistry::new_from_components(vec![]);
        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id.clone(),
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            join_next_blocking_strategy,
            fn_registry.clone(),
        )
        .await;

        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let child_execution_id = execution_id.next_level(&join_set_id);

        apply_create_join_set_start_async(
            &mut caching_db_connection,
            &mut event_history,
            join_set_id.clone(),
            MOCK_FFQN,
            child_execution_id.clone(),
            sim_clock.now(),
        )
        .await;

        // append child response before issuing join_next
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: child_execution_id.clone(),
                        finished_version: Version(0), // does not matter
                        result: CHILD_RESP,
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");

        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id,
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            join_next_blocking_strategy,
            fn_registry,
        )
        .await;

        apply_create_join_set_start_async(
            &mut caching_db_connection,
            &mut event_history,
            join_set_id.clone(),
            MOCK_FFQN,
            child_execution_id.clone(),
            sim_clock.now(),
        )
        .await;
        // issue BlockingChildJoinNext
        let res = event_history
            .apply(
                EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                    JoinNextRequestingFfqn {
                        join_set_id,
                        wasm_backtrace: None,
                        requested_ffqn: MOCK_FFQN,
                    },
                )),
                &mut caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();

        let expected_child_res = WastVal::Result(Ok(Some(Box::new(WastVal::U8(1)))));
        let res = assert_matches!(res, ChildReturnValue::JoinNextRequestingFfqn(Ok(res)) => res);
        assert_eq!((child_execution_id, expected_child_res), res);

        event_history
            .finalize(&mut caching_db_connection, sim_clock.now())
            .await
            .unwrap();

        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn create_two_non_blocking_childs_then_two_join_nexts(
        #[values(true, false)] submits_and_awaits_in_correct_order: bool,
    ) {
        const KID_A_RET: SupportedFunctionReturnValue = SupportedFunctionReturnValue::Ok {
            ok: Some(WastValWithType {
                r#type: TypeWrapper::U8,
                value: WastVal::U8(1),
            }),
        };
        const KID_B_RET: SupportedFunctionReturnValue = SupportedFunctionReturnValue::Ok {
            ok: Some(WastValWithType {
                r#type: TypeWrapper::U8,
                value: WastVal::U8(2),
            }),
        };

        test_utils::set_up();

        let submit_ffqn_1 = if submits_and_awaits_in_correct_order {
            MOCK_FFQN
        } else {
            MOCK_FFQN_2
        };
        let submit_ffqn_2 = if submits_and_awaits_in_correct_order {
            MOCK_FFQN_2
        } else {
            MOCK_FFQN
        };

        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();

        // Create an execution.
        let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;
        let fn_registry = TestingFnRegistry::new_from_components(vec![]);
        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id.clone(),
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            JoinNextBlockingStrategy::Interrupt, // First blocking strategy is always Interrupt
            fn_registry.clone(),
        )
        .await;

        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::Generated, StrVariant::empty()).unwrap();

        let child_execution_id_a = execution_id.next_level(&join_set_id);
        let child_execution_id_b = child_execution_id_a.get_incremented();

        // persist create_join_set, 2x start_async, 1x await_next
        assert_matches!(
            apply_create_join_set_two_start_asyncs_await_next_a(
                &mut caching_db_connection,
                &mut event_history,
                join_set_id.clone(),
                submit_ffqn_1.clone(),
                child_execution_id_a.clone(),
                submit_ffqn_2.clone(),
                child_execution_id_b.clone(),
                sim_clock.now()
            )
            .await
            .unwrap_err(),
            ApplyError::InterruptDbUpdated
        );
        // append two responses
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: if submits_and_awaits_in_correct_order {
                            child_execution_id_a.clone()
                        } else {
                            child_execution_id_b.clone()
                        },
                        finished_version: Version(0), // does not matter
                        result: KID_A_RET,            // won't matter on mismatch
                    },
                },
            )
            .await
            .unwrap();
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: if submits_and_awaits_in_correct_order {
                            child_execution_id_b.clone()
                        } else {
                            child_execution_id_a.clone()
                        },
                        finished_version: Version(0), // does not matter
                        result: KID_B_RET,            // won't matter on mismatch
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");

        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id,
            sim_clock.now(),
            Duration::ZERO, // deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            JoinNextBlockingStrategy::Interrupt,
            fn_registry,
        )
        .await;

        let res = apply_create_join_set_two_start_asyncs_await_next_a(
            &mut caching_db_connection,
            &mut event_history,
            join_set_id.clone(),
            submit_ffqn_1.clone(),
            child_execution_id_a.clone(),
            submit_ffqn_2.clone(),
            child_execution_id_b.clone(),
            sim_clock.now(),
        )
        .await
        .unwrap();
        if !submits_and_awaits_in_correct_order {
            let err = res.unwrap_err();
            // actually execution B was processed.
            assert_eq!(
                AwaitNextExtensionError::FunctionMismatch {
                    specified_function: submit_ffqn_1,
                    actual_function: Some(submit_ffqn_2),
                    actual_id: ResponseId::ChildExecutionId(child_execution_id_b)
                },
                err
            );
        } else {
            let ok = res.unwrap();
            let expected_kid_a_res = WastVal::Result(Ok(Some(Box::new(WastVal::U8(1)))));
            assert_eq!((child_execution_id_a, expected_kid_a_res), ok);
            // second child result should be found
            let res = event_history
                .apply(
                    EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                        JoinNextRequestingFfqn {
                            join_set_id,
                            wasm_backtrace: None,
                            requested_ffqn: submit_ffqn_2.clone(),
                        },
                    )),
                    &mut caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();
            let expected_kid_b_res = WastVal::Result(Ok(Some(Box::new(WastVal::U8(2)))));
            let res =
                assert_matches!(res, ChildReturnValue::JoinNextRequestingFfqn(Ok(res)) => res);
            assert_eq!((child_execution_id_b, expected_kid_b_res), res);
        }

        event_history
            .finalize(&mut caching_db_connection, sim_clock.now())
            .await
            .unwrap();
        db_close.close().await;
    }

    #[tokio::test]
    async fn schedule_event_should_be_processed() {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();
        let db_connection = db_connection.as_ref();

        // Create an execution.
        let execution_id = create_execution(db_connection, &sim_clock).await;
        let fn_registry = TestingFnRegistry::new_from_components(vec![]);
        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id.clone(),
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            JoinNextBlockingStrategy::Interrupt, // does not matter, there are no blocking events
            fn_registry,
        )
        .await;

        event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::Schedule(Schedule {
                    schedule_at: HistoryEventScheduleAt::Now,
                    scheduled_at_if_new: sim_clock.now(),
                    execution_id: ExecutionId::generate(),
                    ffqn: MOCK_FFQN,
                    fn_component_id: ComponentId::dummy_activity(),
                    params: Params::empty(),
                    wasm_backtrace: None,
                })),
                &mut caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();
        // Create a join set to force checking of "Processed" status of the schedule event.
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::JoinSetCreate(JoinSetCreate {
                    join_set_id: join_set_id.clone(),
                    wasm_backtrace: None,
                })),
                &mut caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();

        event_history
            .finalize(&mut caching_db_connection, sim_clock.now())
            .await
            .unwrap();
        db_close.close().await;
    }

    #[tokio::test]
    async fn submit_stub_await() {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();
        let db_connection = db_connection.as_ref();

        let execution_id = create_execution(db_connection, &sim_clock).await;
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let child_execution_id = execution_id.next_level(&join_set_id);
        let fn_registry = TestingFnRegistry::new_from_components(vec![]);
        for run_id in 0..=1 {
            info!("Run {run_id}");
            let (mut event_history, mut caching_db_connection) = load_event_history(
                db_pool.connection(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(sim_clock.clone()),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: 0,
                },
                fn_registry.clone(),
            )
            .await;

            apply_create_join_set_start_async(
                &mut caching_db_connection,
                &mut event_history,
                join_set_id.clone(),
                MOCK_FFQN,
                child_execution_id.clone(),
                sim_clock.now(),
            )
            .await;

            event_history
                .apply(
                    EventCall::NonBlocking(EventCallNonBlocking::Stub(Stub {
                        target_ffqn: MOCK_FFQN,
                        target_execution_id: child_execution_id.clone(),
                        parent_id: execution_id.clone(),
                        join_set_id: join_set_id.clone(),
                        result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                        wasm_backtrace: None,
                    })),
                    &mut caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();

            let child_return_value = event_history
                .apply(
                    EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                        JoinNextRequestingFfqn {
                            join_set_id: join_set_id.clone(),
                            wasm_backtrace: None,
                            requested_ffqn: MOCK_FFQN,
                        },
                    )),
                    &mut caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();

            let res = assert_matches!(
                child_return_value,
                ChildReturnValue::JoinNextRequestingFfqn(Ok(res)) => res
            );
            assert_eq!(
                (
                    child_execution_id.clone(),
                    SUPPORTED_RETURN_VALUE_OK_EMPTY.into_wast_val(|| unreachable!())
                ),
                res
            );

            event_history
                .finalize(&mut caching_db_connection, sim_clock.now())
                .await
                .unwrap();
        }

        db_close.close().await;
    }

    #[tokio::test]
    // Two executions are setting the same stub value to the target activity_stub.
    async fn stubbing_many_times_with_same_value_should_be_ok() {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();
        let db_connection = db_connection.as_ref();

        let execution_id = create_execution(db_connection, &sim_clock).await;
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let target_activity_stub = execution_id.next_level(&join_set_id);
        let fn_registry = TestingFnRegistry::new_from_components(vec![]);
        // First execution creates `target_activity_stub` and stubs its return value.
        for run_id in 0..=1 {
            info!("Run {run_id}");
            let (mut event_history, mut caching_db_connection) = load_event_history(
                db_pool.connection(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1),
                deadline_tracker_factory_test(sim_clock.clone()),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: 0,
                },
                fn_registry.clone(),
            )
            .await;
            apply_create_join_set_start_async(
                &mut caching_db_connection,
                &mut event_history,
                join_set_id.clone(),
                MOCK_FFQN,
                target_activity_stub.clone(),
                sim_clock.now(),
            )
            .await;
            for _ in 0..=1 {
                // In the same execution we can ask many times to stub the same value to target_activity_stub
                event_history
                    .apply(
                        EventCall::NonBlocking(EventCallNonBlocking::Stub(Stub {
                            target_ffqn: MOCK_FFQN,
                            target_execution_id: target_activity_stub.clone(),
                            parent_id: execution_id.clone(),
                            join_set_id: join_set_id.clone(),
                            result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                            wasm_backtrace: None,
                        })),
                        &mut caching_db_connection,
                        sim_clock.now(),
                    )
                    .await
                    .unwrap();
            }

            event_history
                .finalize(&mut caching_db_connection, sim_clock.now())
                .await
                .unwrap();
        }
        drop(execution_id);
        // Second execution
        {
            let execution_id = create_execution(db_connection, &sim_clock).await;
            let (mut event_history, mut caching_db_connection) = load_event_history(
                db_pool.connection(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1),
                deadline_tracker_factory_test(sim_clock.clone()),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: 0,
                },
                fn_registry.clone(),
            )
            .await;
            for _ in 0..=1 {
                // In the same execution we can ask many times to stub the same value to target_activity_stub
                event_history
                    .apply(
                        EventCall::NonBlocking(EventCallNonBlocking::Stub(Stub {
                            target_ffqn: MOCK_FFQN,
                            target_execution_id: target_activity_stub.clone(),
                            parent_id: execution_id.clone(),
                            join_set_id: join_set_id.clone(),
                            result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                            wasm_backtrace: None,
                        })),
                        &mut caching_db_connection,
                        sim_clock.now(),
                    )
                    .await
                    .unwrap();
            }

            event_history
                .finalize(&mut caching_db_connection, sim_clock.now())
                .await
                .unwrap();
        }

        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn trimmed_second_execution_should_result_in_nondeterminism_detected(
        #[values(JoinNextBlockingStrategy::Interrupt, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 0}, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 100})]
        second_run_strategy: JoinNextBlockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, _db_exec, db_close) = Database::Memory.set_up().await;
        let db_connection = db_pool.connection();

        // Create an execution.
        let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;

        let fn_registry = TestingFnRegistry::new_from_components(vec![]);

        let (mut event_history, mut caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id.clone(),
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            JoinNextBlockingStrategy::Interrupt, // first run needs to interrupt
            fn_registry.clone(),
        )
        .await;

        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        let child_execution_id = execution_id.next_level(&join_set_id);

        assert_matches!(
            apply_create_join_set_start_async_await_next(
                &mut caching_db_connection,
                MOCK_FFQN,
                child_execution_id.clone(),
                &mut event_history,
                join_set_id.clone(),
                sim_clock.now()
            )
            .await
            .unwrap_err(),
            ApplyError::InterruptDbUpdated,
            "should have ended with an interrupt"
        );
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: child_execution_id.clone(),
                        finished_version: Version(0), // does not matter
                        result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run attemts to finish with no requests");
        let (mut event_history, _caching_db_connection) = load_event_history(
            db_pool.connection(),
            execution_id,
            sim_clock.now(),
            Duration::from_secs(1), // execution deadline
            deadline_tracker_factory_test(sim_clock.clone()),
            second_run_strategy,
            fn_registry,
        )
        .await;

        // finish the execution
        let err = event_history
            .finalize(&mut caching_db_connection, sim_clock.now())
            .await
            .unwrap_err();
        let reason = assert_matches!(err, ApplyError::NondeterminismDetected(reason) => reason);
        assert_eq!(
            "found unprocessed request stored at version 1: event: JoinSetCreate(o:)",
            reason
        );

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

    // TODO: Check -await-next for fn without return type
    // TODO: Check execution errors translating to await-next-extension-error

    // utils

    async fn create_execution(
        db_connection: &dyn DbConnection,
        sim_clock: &SimClock,
    ) -> ExecutionId {
        let created_at = sim_clock.now();
        let execution_id = ExecutionId::generate();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: MOCK_FFQN,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: ComponentId::dummy_activity(),
                scheduled_by: None,
            })
            .await
            .unwrap();
        execution_id
    }

    async fn load_event_history(
        db_connection: Box<dyn DbConnection>,
        execution_id: ExecutionId,
        now: DateTime<Utc>,
        lock_expires_at: Duration,
        deadline_factory: Arc<dyn DeadlineTrackerFactory>,
        join_next_blocking_strategy: JoinNextBlockingStrategy,
        fn_registry: Arc<dyn FunctionRegistry>,
    ) -> (EventHistory, CachingDbConnection) {
        let execution_deadline = now + lock_expires_at;
        let deadline_tracker = deadline_factory.create(execution_deadline).unwrap();

        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let caching_db_connection = CachingDbConnection {
            db_connection,
            execution_id,
            caching_buffer: CachingBuffer::new(join_next_blocking_strategy),
            version: exec_log.next_version.clone(),
        };
        let cancel_registry = CancelRegistry::new();
        let event_history = EventHistory::new(
            exec_log.event_history().collect(),
            exec_log
                .responses
                .into_iter()
                .map(|event| event.event)
                .collect(),
            join_next_blocking_strategy,
            fn_registry,
            cancel_registry,
            deadline_tracker,
            Locked {
                component_id: ComponentId::dummy_activity(),
                executor_id: ExecutorId::generate(),
                run_id: RunId::generate(),
                lock_expires_at: execution_deadline,
                retry_config: ComponentRetryConfig::ZERO,
            },
            Duration::ZERO, /* lock_extension */
            info_span!("worker-test"),
        );

        (event_history, caching_db_connection)
    }

    async fn apply_create_join_set_start_async_await_next(
        db_connection: &mut CachingDbConnection,
        ffqn: FunctionFqn,
        child_execution_id: ExecutionIdDerived,
        event_history: &mut EventHistory,
        join_set_id: JoinSetId,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, ApplyError> {
        apply_create_join_set_start_async(
            db_connection,
            event_history,
            join_set_id.clone(),
            ffqn.clone(),
            child_execution_id,
            called_at,
        )
        .await;
        event_history
            .apply_inner(
                EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                    JoinNextRequestingFfqn {
                        join_set_id,
                        wasm_backtrace: None,
                        requested_ffqn: ffqn,
                    },
                )),
                db_connection,
                called_at,
            )
            .await
    }

    async fn apply_create_join_set_start_async(
        db_connection: &mut CachingDbConnection,
        event_history: &mut EventHistory,
        join_set_id: JoinSetId,
        ffqn: FunctionFqn,
        child_execution_id: ExecutionIdDerived,
        called_at: DateTime<Utc>,
    ) {
        event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::JoinSetCreate(JoinSetCreate {
                    join_set_id: join_set_id.clone(),
                    wasm_backtrace: None,
                })),
                db_connection,
                called_at,
            )
            .await
            .unwrap();
        event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::SubmitChildExecution(
                    SubmitChildExecution {
                        target_ffqn: ffqn,
                        fn_component_id: ComponentId::dummy_activity(),
                        join_set_id,
                        child_execution_id,
                        params: Params::empty(),
                        wasm_backtrace: None,
                    },
                )),
                db_connection,
                called_at,
            )
            .await
            .unwrap();
    }

    #[expect(clippy::too_many_arguments)]
    async fn apply_create_join_set_two_start_asyncs_await_next_a(
        db_connection: &mut CachingDbConnection,
        event_history: &mut EventHistory,
        join_set_id: JoinSetId,
        ffqn_a: FunctionFqn,
        child_execution_id_a: ExecutionIdDerived,
        ffqn_b: FunctionFqn,
        child_execution_id_b: ExecutionIdDerived,
        called_at: DateTime<Utc>,
    ) -> Result<Result<(ExecutionIdDerived, WastVal), AwaitNextExtensionError>, ApplyError> {
        apply_create_join_set_start_async(
            db_connection,
            event_history,
            join_set_id.clone(),
            ffqn_a.clone(),
            child_execution_id_a,
            called_at,
        )
        .await;
        event_history
            .apply(
                EventCall::NonBlocking(EventCallNonBlocking::SubmitChildExecution(
                    SubmitChildExecution {
                        target_ffqn: ffqn_b,
                        fn_component_id: ComponentId::dummy_activity(),
                        join_set_id: join_set_id.clone(),
                        child_execution_id: child_execution_id_b,
                        params: Params::empty(),
                        wasm_backtrace: None,
                    },
                )),
                db_connection,
                called_at,
            )
            .await
            .unwrap();
        event_history
            .apply_inner(
                EventCall::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                    JoinNextRequestingFfqn {
                        join_set_id,
                        wasm_backtrace: None,
                        requested_ffqn: ffqn_a,
                    },
                )),
                db_connection,
                called_at,
            )
            .await
            .map(|res| match res {
                ChildReturnValue::JoinNextRequestingFfqn(res) => res,
                other => {
                    unreachable!(
                        "BlockingChildAwaitNext returns JoinNextRequestingFfqn, got {other:?}"
                    )
                }
            })
    }
}