obeli-sk-wasm-workers 0.41.5

use super::caching_db_connection::CacheableDbEvent;
use super::caching_db_connection::WorkflowDbConnection;
use super::deadline_tracker::DeadlineTracker;
use super::event_history::ProcessingStatus::Processed;
use super::event_history::ProcessingStatus::Unprocessed;
use super::host_exports::latest::obelisk::types::execution::GetExtensionError;
use super::workflow_ctx::WorkflowFunctionError;
use super::workflow_worker::JoinNextBlockingStrategy;
use crate::activity::cancel_registry::CancelRegistry;
use crate::workflow::deadline_tracker::{InterruptKind, PreemptRequested};
use crate::workflow::host_exports::ffqn_into_wast_val;
use crate::workflow::host_exports::latest;
use crate::workflow::host_exports::latest::obelisk::types::execution as types_execution;
use crate::workflow::host_exports::latest::obelisk::workflow::workflow_support::JoinNextError;
use crate::workflow::host_exports::latest::obelisk::workflow::workflow_support::JoinNextTryError;
use crate::workflow::replay_advance::JoinSetCloseCancellations;
use assert_matches::assert_matches;
use chrono::{DateTime, Utc};
use concepts::ComponentId;
use concepts::ExecutionMetadata;
use concepts::FunctionRegistry;
use concepts::InvalidNameError;
use concepts::JoinSetId;
use concepts::JoinSetKind;
use concepts::SupportedFunctionReturnValue;
use concepts::prefixed_ulid::ExecutionIdTopLevel;
use concepts::prefixed_ulid::{DelayId, DeploymentId, ExecutionIdDerived};
use concepts::storage;
use concepts::storage::AppendResponseToExecution;
use concepts::storage::BacktraceInfo;
use concepts::storage::ChildExecutionRequestError;
use concepts::storage::DbErrorWrite;
use concepts::storage::HistoryEventScheduleAt;
use concepts::storage::Locked;
use concepts::storage::PersistKind;
use concepts::storage::ResponseCursor;
use concepts::storage::ResponseWithCursor;
use concepts::storage::ScheduleRequestError;
use concepts::storage::StubError;
use concepts::storage::StubRetValHash;
use concepts::storage::{
    AppendRequest, CreateRequest, ExecutionRequest, JoinSetResponse, JoinSetResponseEvent, Version,
};
use concepts::storage::{HistoryEvent, JoinNextTryOutcome, JoinSetRequest};
use concepts::storage::{ResponseSubscriptionEnd, SubscribeToResponsesError};
use concepts::{ExecutionId, StrVariant};
use concepts::{FunctionFqn, Params};
use db_common::{JoinSetOpenTracker, JoinSetOpenTrackerError, JoinSetResponseId};
use hashbrown::HashMap;
use indexmap::IndexMap;
use indexmap::indexmap;
use std::collections::VecDeque;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tracing::Level;
use tracing::Span;
use tracing::info;
use tracing::instrument;
use tracing::{debug, trace};
use val_json::wast_val::ValKey;
use val_json::wast_val::WastVal;
use wasmtime::component::Val;

#[derive(Debug)]
enum ChildReturnValue {
    WastVal(WastVal),
    JoinSetCreate(JoinSetId),
    JoinNext(Result<(JoinSetResponseId, Result<(), ()>), JoinNextError>),
    JoinNextTry(Result<(JoinSetResponseId, Result<(), ()>), JoinNextTryError>),
    JoinNextRequestingFfqn(Result<(ExecutionIdDerived, WastVal), AwaitNextExtensionError>),
    OneOffDelay {
        scheduled_at: DateTime<Utc>,
        result: Result<(), ()>,
    },
    SubmitDelay,
    Stub(Result<(), StubError>),
    Schedule(Result<(), ScheduleRequestError>),
    SubmitChild(Result<(), ChildExecutionRequestError>),
    Persist,
}

struct AppendedBlockingEvents {
    history_events: Vec<(HistoryEvent, Version)>,
    known_response: Option<ChildReturnValue>,
}

#[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(#[from] DbErrorWrite),
    #[error("constraint violation: {0}")]
    ConstraintViolation(StrVariant),
    #[error("execution interrupt: {0:?}")]
    Interrupt(InterruptKind),
    #[error("replay interrupt")]
    ReplayInterrupt,
}

#[derive(Debug, Clone, thiserror::Error)]
pub(crate) enum DbErrorWriteOrReplayInterrupt {
    #[error(transparent)]
    DbError(#[from] DbErrorWrite),
    #[error("replay interrupt")]
    ReplayInterrupt,
}
impl From<DbErrorWriteOrReplayInterrupt> for ApplyError {
    fn from(value: DbErrorWriteOrReplayInterrupt) -> Self {
        match value {
            DbErrorWriteOrReplayInterrupt::DbError(err) => ApplyError::DbError(err),
            DbErrorWriteOrReplayInterrupt::ReplayInterrupt => ApplyError::ReplayInterrupt,
        }
    }
}

fn join_set_open_tracker_error_to_constraint(err: &JoinSetOpenTrackerError) -> StrVariant {
    err.to_string().into()
}

#[derive(Debug, thiserror::Error)]
pub(crate) enum UpsertStubOrReplayInterrupt {
    #[error("stub conflict")]
    StubConflict,
    #[error(transparent)]
    DbError(DbErrorWrite),
    #[error("replay interrupt")]
    ReplayInterrupt,
}

#[expect(clippy::struct_field_names)]
pub(crate) struct EventHistory {
    replaying_unfinished_execution: bool,
    deployment_id: DeploymentId,
    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>,
    index_delay_id_to_expires_at: IndexMap<DelayId, DateTime<Utc>>,
    responses: Vec<(ResponseWithCursor, ProcessingStatus)>,
    worker_span: Span,
    pub(crate) deadline_tracker: Box<dyn DeadlineTracker>,
    lock_extension: Duration, // 0 == disabled
    pub(crate) locked_event: Locked,
    pub(crate) fn_registry: Arc<dyn FunctionRegistry>,
    cancel_registry: CancelRegistry,

    subscription_interruption: Option<Duration>,

    // Tracks join set contents for closing. One-offs are ignored, response id is removed when the response is processed.
    join_set_open_tracker: JoinSetOpenTracker,

    // Id of the last processed response per (non-one-off) join set, for `join-set.last-id`.
    // Rebuilt deterministically on replay as `join-next` / `-await-next` are re-applied.
    last_response_by_join_set: HashMap<JoinSetId, JoinSetResponseId>,
    // Last one-off response of any kind (incl. sleep delay), for `last-oneoff-id`.
    last_oneoff_id: Option<JoinSetResponseId>,
    // Last one-off child execution (call-json / direct call), for `last-direct-call-id`.
    last_direct_call_id: Option<ExecutionIdDerived>,

    // Upper bound on captured writes a replay pass may collect. `None` outside replay.
    max_replay_captured_writes: Option<usize>,
    max_events_per_run: Option<usize>,
    written_events_this_run: usize,
    response_refresh_interval: Option<usize>,
    events_since_response_refresh: usize,
}

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

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

#[derive(Debug)]
enum FindMatchingAtomicResponse {
    Found {
        value: ChildReturnValue,
        version_range: EventCallVersionRange,
    },
    NotFound,
    FoundRequestButNotResponse {
        version_range: EventCallVersionRange,
    },
}

#[derive(Debug, Clone)]
struct EventCallVersionRange {
    min_including: Version,
    max_excluding: Version,
}
impl EventHistory {
    #[expect(clippy::too_many_arguments)]
    pub(crate) fn new(
        deployment_id: DeploymentId,
        event_history: Vec<(HistoryEvent, Version)>,
        responses: Vec<ResponseWithCursor>,
        join_next_blocking_strategy: JoinNextBlockingStrategy,
        fn_registry: Arc<dyn FunctionRegistry>,
        cancel_registry: CancelRegistry,
        deadline_tracker: Box<dyn DeadlineTracker>,
        locked_event: Locked,
        lock_extension: Option<Duration>,
        subscription_interruption: Option<Duration>,
        worker_span: Span,
        replaying_unfinished_execution: bool,
        max_replay_captured_writes: Option<usize>,
        max_events_per_run: Option<usize>,
        response_refresh_interval: Option<usize>,
    ) -> EventHistory {
        EventHistory {
            replaying_unfinished_execution,
            deployment_id,
            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: lock_extension.unwrap_or_default(),
            subscription_interruption,
            join_set_open_tracker: JoinSetOpenTracker::new(),
            last_response_by_join_set: HashMap::default(),
            last_oneoff_id: None,
            last_direct_call_id: None,
            max_replay_captured_writes,
            max_events_per_run,
            written_events_this_run: 0,
            response_refresh_interval,
            events_since_response_refresh: 0,
        }
    }

    /// Record the last processed response id of a (non-one-off) join set for `join-set.last-id`.
    pub(crate) fn record_last_response_id(
        &mut self,
        join_set_id: &JoinSetId,
        response_id: JoinSetResponseId,
    ) {
        self.last_response_by_join_set
            .insert(join_set_id.clone(), response_id);
    }

    /// Record the last one-off response (sleep / call-json / direct call) for
    /// `last-oneoff-id` and, for a child execution, `last-direct-call-id`.
    pub(crate) fn record_last_oneoff_id(&mut self, response_id: JoinSetResponseId) {
        if let JoinSetResponseId::ChildExecutionId(child_id) = &response_id {
            self.last_direct_call_id = Some(child_id.clone());
        }
        self.last_oneoff_id = Some(response_id);
    }

    pub(crate) fn last_response_id(&self, join_set_id: &JoinSetId) -> Option<&JoinSetResponseId> {
        self.last_response_by_join_set.get(join_set_id)
    }

    pub(crate) fn last_oneoff_id(&self) -> Option<&JoinSetResponseId> {
        self.last_oneoff_id.as_ref()
    }

    pub(crate) fn last_direct_call_id(&self) -> Option<&ExecutionIdDerived> {
        self.last_direct_call_id.as_ref()
    }

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

    pub(crate) fn has_unprocessed_responses(&self) -> bool {
        self.responses.iter().any(|(_response, processing_status)| {
            *processing_status == ProcessingStatus::Unprocessed
        })
    }

    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 unwrapped from trap by the worker.
    async fn apply(
        &mut self,
        event_call: EventCallKind,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, WorkflowFunctionError> {
        Ok(self
            .apply_event_call(event_call, event_call_cursor, db_connection, called_at)
            .await?)
    }

    async fn apply_event_call(
        &mut self,
        event_call_kind: EventCallKind,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, ApplyError> {
        match self.deadline_tracker.check_preempt() {
            Ok(()) => {}
            Err(PreemptRequested::Interrupt(kind)) => {
                info!("Execution interrupt detected in check_preempt: {kind:?}");
                return Err(ApplyError::Interrupt(kind));
            }
        }

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

        let event_call = event_call_cursor.next(event_call_kind);
        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(?event_call))]
    async fn apply_inner(
        &mut self,
        event_call: EventCall,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<ChildReturnValue, ApplyError> {
        debug!("applying {event_call:?}");

        // A non-terminating workflow (e.g. a `joinNextTry` busy loop on a response that is never
        // injected) would collect captured writes forever during replay, since the replay deadline
        // never expires the lock. Once the collection hits the bound, interrupt replay so the writes
        // gathered so far are returned as an advanceable prefix; advancing them and replaying again
        // resumes collection from the persisted tip, stepping the workflow forward N writes at a time.
        if let Some(max) = self.max_replay_captured_writes
            && let Some(collected) = db_connection.captured_writes_collected()
            && collected >= max
        {
            debug!("Reached replay captured-write limit of {max}, interrupting replay");
            return Err(ApplyError::ReplayInterrupt);
        }

        let wasm_backtrace = event_call.wasm_backtrace().cloned();
        match self.find_matching_atomic(&event_call)? {
            FindMatchingAtomicResponse::Found {
                value,
                version_range,
            } => {
                trace!("found_atomic: {value:?}");
                if let Some(wasm_backtrace) = wasm_backtrace {
                    self.persist_replayed_backtrace(db_connection, version_range, wasm_backtrace)
                        .await?;
                }
                return Ok(value);
            }
            FindMatchingAtomicResponse::NotFound => {} // continue
            FindMatchingAtomicResponse::FoundRequestButNotResponse { version_range, .. } => {
                assert!(self.replaying_unfinished_execution);
                if let Some(wasm_backtrace) = wasm_backtrace {
                    self.persist_replayed_backtrace(db_connection, version_range, wasm_backtrace)
                        .await?;
                }
                return Err(ApplyError::ReplayInterrupt);
            }
        }

        let version_range = event_call.version_range.clone();
        let event_history_len_before = self.event_history.len();
        let value = match event_call.kind {
            EventCallKind::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,
                        version_range.min_including.clone(),
                        db_connection,
                        called_at,
                    )
                    .await?;
                self.event_history.push((event, Unprocessed, version));
                if self.response_refresh_interval.is_some() {
                    self.events_since_response_refresh += 1;
                }
                let refreshed_responses = if self
                    .response_refresh_interval
                    .is_some_and(|interval| self.events_since_response_refresh >= interval)
                {
                    db_connection
                        .flush_non_blocking_event_cache(called_at)
                        .await?;
                    Some(self.poll_responses(db_connection).await?)
                } else {
                    None
                };
                trace!("find_matching_atomic must mark the non-blocking event as Processed");
                let stored_event_call = EventCall::new(
                    version_range.min_including.clone(),
                    version_range,
                    EventCallKind::NonBlocking(cloned_non_blocking),
                );
                let non_blocking_resp = assert_matches!(
                    self.find_matching_atomic(&stored_event_call)?,
                    FindMatchingAtomicResponse::Found { value, .. } => value, "just stored the event as Unprocessed, it must be found");
                if let Some(refreshed_responses) = refreshed_responses {
                    self.extend_responses(refreshed_responses);
                }
                Ok(non_blocking_resp)
            }
            EventCallKind::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,
                    version_range.min_including,
                    db_connection,
                    lock_expires_at,
                    called_at,
                )
                .await
            }
        }?;
        let written_events = self.event_history.len() - event_history_len_before;
        self.written_events_this_run += written_events;
        if written_events > 0
            && self
                .max_events_per_run
                .is_some_and(|max| self.written_events_this_run >= max)
        {
            return Err(ApplyError::Interrupt(
                InterruptKind::WorkflowEventLimitReached,
            ));
        }
        Ok(value)
    }

    async fn poll_responses(
        &mut self,
        db_connection: &mut dyn WorkflowDbConnection,
    ) -> Result<Vec<ResponseWithCursor>, ApplyError> {
        let last_response = self.last_response_cursor();
        self.events_since_response_refresh = 0;
        match db_connection
            .subscribe_to_next_responses(
                db_connection.execution_id(),
                last_response,
                Box::pin(std::future::ready(
                    ResponseSubscriptionEnd::PollIntervalElapsed,
                )),
            )
            .await
        {
            Ok(responses) => Ok(responses),
            Err(SubscribeToResponsesError::DbErrorRead(err)) => {
                Err(ApplyError::DbError(DbErrorWrite::from(err)))
            }
            Err(SubscribeToResponsesError::SubscriptionEnded(
                ResponseSubscriptionEnd::PollIntervalElapsed
                | ResponseSubscriptionEnd::LockDeadlineReached,
            )) => Ok(Vec::new()),
            Err(SubscribeToResponsesError::SubscriptionEnded(
                ResponseSubscriptionEnd::ExecutorClosing,
            )) => Err(ApplyError::Interrupt(InterruptKind::ExecutorClosing)),
            Err(SubscribeToResponsesError::SubscriptionEnded(
                ResponseSubscriptionEnd::ExecutionUpdated,
            )) => Err(ApplyError::Interrupt(InterruptKind::PauseOrCancel)),
        }
    }

    fn last_response_cursor(&self) -> ResponseCursor {
        self.responses
            .last()
            .map(|(resp, _)| resp.cursor)
            .unwrap_or(ResponseCursor(0))
    }

    fn extend_responses(&mut self, next_responses: Vec<ResponseWithCursor>) {
        trace!("Got next responses {next_responses:?}");
        self.responses
            .extend(next_responses.into_iter().map(|resp| (resp, Unprocessed)));
        trace!("All responses: {:?}", self.responses);
    }

    async fn persist_replayed_backtrace(
        &self,
        db_connection: &mut dyn WorkflowDbConnection,
        version_range: EventCallVersionRange,
        wasm_backtrace: storage::WasmBacktrace,
    ) -> Result<(), DbErrorWrite> {
        let backtrace = BacktraceInfo {
            execution_id: db_connection.execution_id().clone(),
            component_id: self.locked_event.component_id.clone(),
            version_min_including: version_range.min_including,
            version_max_excluding: version_range.max_excluding,
            wasm_backtrace,
        };
        db_connection.append_backtrace(backtrace).await
    }

    async fn extend_lock(
        &mut self,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        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: ExecutionRequest::Locked(self.locked_event.clone()),
        };
        info!(
            "Extending the lock at version {version}",
            version = event_call_cursor.version()
        );

        let version = event_call_cursor.next_version.clone();
        db_connection
            .append_blocking(
                version,
                db_connection.execution_id().clone(),
                append_req,
                None,
                &self.locked_event.component_id,
            )
            .await?;
        event_call_cursor.next_version = event_call_cursor.next_version.increment();
        Ok(())
    }

    async fn apply_blocking(
        &mut self,
        event_call: EventCallBlocking,
        version: Version,
        db_connection: &mut dyn WorkflowDbConnection,
        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 AppendedBlockingEvents {
            history_events,
            mut known_response,
        } = self
            .append_to_db_blocking(
                event_call,
                version,
                db_connection,
                called_at,
                lock_expires_at,
            )
            .await?;
        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 response = self.process_event_by_key(&key)?;
            if idx == last_key_idx {
                match (response, known_response.take()) {
                    (FindMatchingResponse::Found { value, .. }, None) => {
                        assert_eq!(
                            Processed,
                            self.event_history
                                .last()
                                .expect("checked that `history_events` is not empty")
                                .1
                        );
                        return Ok(value);
                    }
                    (FindMatchingResponse::FoundRequestButNotResponse { .. }, Some(value)) => {
                        self.event_history
                            .last_mut()
                            .expect("checked that `history_events` is not empty")
                            .1 = Processed;
                        return Ok(value);
                    }
                    (_, Some(_)) => {
                        unreachable!("a locally resolved blocking event must await its response")
                    }
                    _ => {}
                }
            }
        }
        // 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, waking on an interrupt or the deadline.
            loop {
                let subscription_end_fut =
                    match self.deadline_tracker.track(self.subscription_interruption) {
                        Ok(subscription_end_fut) => subscription_end_fut,
                        Err(ResponseSubscriptionEnd::PollIntervalElapsed) => continue,
                        Err(ResponseSubscriptionEnd::LockDeadlineReached) => break,
                        Err(ResponseSubscriptionEnd::ExecutorClosing) => {
                            return Err(ApplyError::Interrupt(InterruptKind::ExecutorClosing));
                        }
                        Err(ResponseSubscriptionEnd::ExecutionUpdated) => {
                            return Err(ApplyError::Interrupt(InterruptKind::PauseOrCancel));
                        }
                    };
                let last_response = self.last_response_cursor();
                self.events_since_response_refresh = 0;
                match db_connection
                    .subscribe_to_next_responses(
                        db_connection.execution_id(),
                        last_response,
                        subscription_end_fut,
                    )
                    .await
                {
                    Ok(next_responses) => {
                        debug!(
                            "Original {orig_len} responses are extended by {len} after old last rep {last_response}, next first: {first:?}, last: {last:?}",
                            orig_len = self.responses.len(),
                            len = next_responses.len(),
                            first = next_responses.first(),
                            last = next_responses.last(),
                        );
                        self.extend_responses(next_responses);
                        if let FindMatchingResponse::Found {
                            value: accept_resp, ..
                        } = self.process_event_by_key(&key)?
                        {
                            debug!(join_set_id = %join_next_variant.join_set_id(), "Got result");
                            return Ok(accept_resp);
                        } // this did not unblock the execution, loop
                    }
                    Err(SubscribeToResponsesError::DbErrorRead(err)) => {
                        return Err(ApplyError::DbError(DbErrorWrite::from(err)));
                    }
                    Err(SubscribeToResponsesError::SubscriptionEnded(
                        ResponseSubscriptionEnd::PollIntervalElapsed,
                    )) => {}
                    Err(SubscribeToResponsesError::SubscriptionEnded(
                        ResponseSubscriptionEnd::LockDeadlineReached,
                    )) => break,
                    Err(SubscribeToResponsesError::SubscriptionEnded(
                        ResponseSubscriptionEnd::ExecutorClosing,
                    )) => return Err(ApplyError::Interrupt(InterruptKind::ExecutorClosing)),
                    Err(SubscribeToResponsesError::SubscriptionEnded(
                        ResponseSubscriptionEnd::ExecutionUpdated,
                    )) => return Err(ApplyError::Interrupt(InterruptKind::PauseOrCancel)),
                }
            }
            debug!("Giving up on waiting for response");
        }
        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,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<(), WorkflowFunctionError> {
        self.join_set_close_inner(
            join_set_id,
            event_call_cursor,
            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,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<(), ApplyError> {
        let response_ids = self
            .join_set_open_tracker
            .close_join_set(join_set_id)
            .map_err(|err| {
                ApplyError::ConstraintViolation(join_set_open_tracker_error_to_constraint(&err))
            })?;
        debug!("Closing `{join_set_id}` with unawaited {response_ids:?}");

        let join_next_count = response_ids.len();
        // Activities+delays are cancelled, cancellable children are signalled, other
        // children are merely awaited. Order retained (cancelled in reverse later).
        let mut activity_and_delay_ids = Vec::new();
        let mut cancellable_child_ids = Vec::new();
        for (response_id, member) in response_ids {
            match &response_id {
                JoinSetResponseId::DelayId(_) => activity_and_delay_ids.push(response_id),
                JoinSetResponseId::ChildExecutionId(child_id) => {
                    if member.is_activity() {
                        activity_and_delay_ids.push(response_id);
                    } else if member.is_cancellable_workflow() {
                        cancellable_child_ids.push(child_id.clone());
                    }
                }
            }
        }
        let mut cancellations =
            if activity_and_delay_ids.is_empty() && cancellable_child_ids.is_empty() {
                None
            } else {
                Some(JoinSetCloseCancellations::new(
                    activity_and_delay_ids,
                    cancellable_child_ids,
                    called_at,
                ))
            };
        for _ in 0..join_next_count {
            let event_call =
                EventCallKind::Blocking(EventCallBlocking::JoinSetClose(JoinSetClose {
                    join_set_id: join_set_id.clone(),
                    cancellations: std::mem::take(&mut cancellations), // First iterations takes all cancellations, so that the activities get the signal ASAP.
                    wasm_backtrace: wasm_backtrace.clone(),
                }));
            self.apply_event_call(event_call, event_call_cursor, db_connection, called_at)
                .await?;
        }
        Ok(())
    }

    /// Close join sets and check determinism.
    /// There must be no unprocessed requests, otherwise the code is trimming previous run.
    #[instrument(skip_all)]
    pub(crate) async fn finalize(
        &mut self,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), ApplyError> {
        while let Some(join_set_id) = self
            .join_set_open_tracker
            .open_join_sets()
            .iter()
            .rev()
            .find_map(|(js, remaining)| {
                if !remaining.is_empty() {
                    Some(js.clone())
                } else {
                    None
                }
            })
        {
            self.join_set_close_inner(
                &join_set_id,
                event_call_cursor,
                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<FindMatchingAtomicResponse, 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() {
            match self.process_event_by_key(&key)? {
                FindMatchingResponse::NotFound => {
                    assert_eq!(idx, 0, "NotFound must be returned on the first key");
                    return Ok(FindMatchingAtomicResponse::NotFound);
                }
                FindMatchingResponse::FoundRequestButNotResponse {
                    version: matched_version,
                } => {
                    let expected_version = Version::new(
                        event_call.version.0
                            + u32::try_from(idx).expect("an EventCall has at most three events"),
                    );
                    assert_eq!(expected_version, matched_version);
                    // JoinNext is still unprocessed. This can only happen on the last key (the JoinNext key).
                    assert_eq!(
                        last_key_idx, idx,
                        "FoundRequestButNotResponse must be returned on the last key"
                    );
                    return Ok(FindMatchingAtomicResponse::FoundRequestButNotResponse {
                        version_range: event_call.version_range.clone(),
                    });
                }
                FindMatchingResponse::Found {
                    value: found,
                    version: matched_version,
                } => {
                    let expected_version = Version::new(
                        event_call.version.0
                            + u32::try_from(idx).expect("an EventCall has at most three events"),
                    );
                    assert_eq!(expected_version, matched_version);
                    if idx == last_key_idx {
                        return Ok(FindMatchingAtomicResponse::Found {
                            value: found,
                            version_range: event_call.version_range.clone(),
                        });
                    }
                }
            }
        }
        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.clone()))
                } else {
                    None
                }
            })
    }

    /// Check if there's an unprocessed response for the given join set.
    fn has_unprocessed_response_for_join_set(&self, join_set_id: &JoinSetId) -> bool {
        self.responses.iter().any(|(event, status)| {
            *status == Unprocessed && event.event.event.join_set_id == *join_set_id
        })
    }

    // Only called from `process_event_by_key`
    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.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.event.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
        }
    }

    // Only entry point to marking events as processed.
    fn process_event_by_key(
        &mut self,
        key: &DeterministicKey,
    ) -> Result<FindMatchingResponse, ApplyError> {
        let Some((found_idx, found_request_event, found_version)) =
            self.first_unprocessed_request()
        else {
            return Ok(FindMatchingResponse::NotFound);
        };
        trace!(
            "Finding match for {key:?}, [{found_idx}, v{found_version}] {found_request_event:?}"
        );
        let response = {
            use ProcessEventResponse as FindMatchingResponse;

            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;
                    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;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::Persist))
                }

                (
                    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,
                                result: found_result,
                            },
                    },
                ) 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, result: {found_result:?}");
                    let found_result = found_result.clone();
                    // Only add to index if the result was Ok (child was actually created)
                    if found_result.is_ok() {
                        self.index_child_exe_to_ffqn
                            .insert(child_execution_id.clone(), target_ffqn.clone());
                    }
                    self.event_history[found_idx].1 = Processed;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::SubmitChild(
                        found_result,
                    )))
                }

                (
                    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 must be 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: JoinSetResponseId::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: JoinSetResponseId::DelayId(delay_id.clone()),
                            };
                            Ok(FindMatchingResponse::Found(
                                ChildReturnValue::JoinNextRequestingFfqn(Err(function_mismatch)),
                            ))
                        }
                        None => Ok(FindMatchingResponse::FoundRequestButNotResponse),
                    }
                }

                (
                    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),
                        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(
                                (
                                    JoinSetResponseId::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(
                                (JoinSetResponseId::DelayId(delay_id.clone()), result),
                            ))))
                        }
                        None => Ok(FindMatchingResponse::FoundRequestButNotResponse),
                    }
                }

                (
                    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(JoinNextError::AllProcessed),
                    )))
                }

                // JoinNextTry with outcome=Found: match with actual response
                (
                    DeterministicKey::JoinNextTry { join_set_id },
                    HistoryEvent::JoinNextTry {
                        join_set_id: found_join_set_id,
                        outcome: JoinNextTryOutcome::Found,
                    },
                ) if *join_set_id == *found_join_set_id => {
                    trace!(%join_set_id, "DeterministicKey::JoinNextTry(found): Peeked at JoinNextTry");
                    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::JoinNextTry: Matched ChildExecutionFinished");
                            Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNextTry(
                                Ok((
                                    JoinSetResponseId::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::JoinNextTry: Matched DelayFinished");
                            Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNextTry(
                                Ok((JoinSetResponseId::DelayId(delay_id.clone()), result)),
                            )))
                        }
                        None => {
                            // This is a nondeterminism: history says outcome=Found but no response available
                            Err(ApplyError::NondeterminismDetected(format!(
                                "JoinNextTry recorded outcome=Found but no response available for join set `{join_set_id}`"
                            )))
                        }
                    }
                }

                // JoinNextTry with outcome=Pending: return Pending error
                (
                    DeterministicKey::JoinNextTry { join_set_id },
                    HistoryEvent::JoinNextTry {
                        join_set_id: found_join_set_id,
                        outcome: JoinNextTryOutcome::Pending,
                    },
                ) if *join_set_id == *found_join_set_id => {
                    trace!(%join_set_id, "DeterministicKey::JoinNextTry(pending): returning Pending error");
                    self.event_history[found_idx].1 = Processed;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNextTry(
                        Err(JoinNextTryError::Pending),
                    )))
                }

                // JoinNextTry with outcome=AllProcessed: return AllProcessed error
                (
                    DeterministicKey::JoinNextTry { join_set_id },
                    HistoryEvent::JoinNextTry {
                        join_set_id: found_join_set_id,
                        outcome: JoinNextTryOutcome::AllProcessed,
                    },
                ) if *join_set_id == *found_join_set_id => {
                    trace!(%join_set_id, "DeterministicKey::JoinNextTry(all_processed): returning AllProcessed error");
                    self.event_history[found_idx].1 = Processed;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::JoinNextTry(
                        Err(JoinNextTryError::AllProcessed),
                    )))
                }

                (
                    DeterministicKey::Schedule {
                        target_execution_id,
                        schedule_at,
                    },
                    HistoryEvent::Schedule {
                        execution_id: found_execution_id,
                        schedule_at: found_schedule_at,
                        result: found_result,
                    },
                ) if *target_execution_id == *found_execution_id
                    && schedule_at == found_schedule_at =>
                {
                    trace!(%target_execution_id, "Matched Schedule, result: {:?}", found_result);
                    // Clone the result before mutating self
                    let found_result = found_result.clone();
                    self.event_history[found_idx].1 = Processed;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::Schedule(
                        found_result,
                    )))
                }

                (
                    DeterministicKey::Stub { intent, params },
                    HistoryEvent::Stub {
                        target_execution_id: found_execution_id,
                        retval_hash: found_retval_hash,
                        result: found_result,
                    },
                ) if params.target_execution_id == *found_execution_id
                    && params.retval_hash == *found_retval_hash =>
                {
                    trace!(target_execution_id = %params.target_execution_id, "Matched Stub");
                    let found_result = found_result.clone();
                    self.event_history[found_idx].1 = Processed;
                    Ok(FindMatchingResponse::Found(ChildReturnValue::Stub(
                        found_result,
                    )))
                }

                (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}",
                    )))
                }
            }
        }?;
        Ok(match response {
            ProcessEventResponse::Found(value) => FindMatchingResponse::Found {
                value,
                version: found_version,
            },
            ProcessEventResponse::FoundRequestButNotResponse => {
                FindMatchingResponse::FoundRequestButNotResponse {
                    version: found_version,
                }
            }
        })
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_to_db_non_blocking(
        &mut self,
        event_call: EventCallNonBlocking,
        version: Version,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(HistoryEvent, Version), DbErrorWriteOrReplayInterrupt> {
        trace!("append_to_db_non_blocking {}", 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(), version.clone());
                let join_set_create = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::JoinSetCreate {
                    request: join_set_create,
                    version: 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: version.clone(),
                        version_max_excluding: Version::new(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(), version.clone());
                let request = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::Persist {
                    request,
                    version: 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: version.clone(),
                        version_max_excluding: Version::new(version.0 + 1),
                    }),
                };
                db_connection
                    .append_non_blocking(cacheable_event, called_at)
                    .await?;
                Ok(history_event)
            }

            EventCallNonBlocking::SubmitChildExecution(SubmitChildExecution {
                target_ffqn,
                join_set_id,
                child_execution_id,
                intent,
                wasm_backtrace,
            }) => {
                // Handle intent: create result and optionally child_req based on intent
                let (result, params, maybe_child_req) = match intent {
                    SubmitChildIntent::Ok {
                        fn_component_id,
                        params,
                    } => {
                        let child_req = CreateRequest {
                            created_at: called_at,
                            execution_id: ExecutionId::Derived(child_execution_id.clone()),
                            ffqn: target_ffqn.clone(),
                            params: params.clone(),
                            parent: Some((
                                db_connection.execution_id().clone(),
                                join_set_id.clone(),
                            )),
                            metadata: ExecutionMetadata::from_parent_span(&self.worker_span),
                            scheduled_at: called_at,
                            component_id: fn_component_id,
                            deployment_id: self.deployment_id,
                            scheduled_by: None,
                            paused: false,
                        };
                        (Ok(()), params, Some(child_req))
                    }
                    SubmitChildIntent::Err(err) => {
                        // Function was not found or params parsing error
                        (Err(err), Params::empty(), None)
                    }
                };

                debug!(%child_execution_id, %join_set_id, "SubmitChildExecution: appending, has_child_req: {}", maybe_child_req.is_some());
                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,
                        result,
                    },
                };
                let history_event = (event.clone(), version.clone());
                let append_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };

                let 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: version.clone(),
                    version_max_excluding: Version::new(version.0 + 1),
                });

                // If we have a child_req, use SubmitChildExecution event; otherwise just persist the history event
                if let Some(child_req) = maybe_child_req {
                    let cacheable_event = CacheableDbEvent::SubmitChildExecution {
                        request: append_req,
                        version: version.clone(),
                        child_req,
                        backtrace,
                    };
                    db_connection
                        .append_non_blocking(cacheable_event, called_at)
                        .await?;
                } else {
                    // Error case: only persist the history event, no child creation
                    let cacheable_event = CacheableDbEvent::SubmitChildExecutionError {
                        request: append_req,
                        version: version.clone(),
                        backtrace,
                    };
                    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,
                        paused: false,
                    },
                };
                let history_event = (event.clone(), version.clone());
                let delay_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                let cacheable_event = CacheableDbEvent::SubmitDelay {
                    request: delay_req,
                    version: 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: version.clone(),
                        version_max_excluding: Version::new(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,
                intent,
                wasm_backtrace,
            }) => {
                // Handle intent: create result and optionally child_req based on intent
                let (result, maybe_child_req) = match intent {
                    ScheduleIntent::Ok {
                        fn_component_id,
                        params,
                    } => {
                        let child_req = CreateRequest {
                            created_at: called_at,
                            execution_id: new_execution_id.clone(),
                            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,
                            deployment_id: self.deployment_id,
                            scheduled_by: Some(db_connection.execution_id().clone()),
                            paused: false,
                        };
                        (Ok(()), Some(child_req))
                    }
                    ScheduleIntent::Err(err) => (Err(err), None),
                };

                let event = HistoryEvent::Schedule {
                    execution_id: new_execution_id.clone(),
                    schedule_at,
                    result,
                };

                let history_event = (event.clone(), version.clone());
                let append_req = AppendRequest {
                    event: ExecutionRequest::HistoryEvent { event },
                    created_at: called_at,
                };
                debug!(%new_execution_id, "ScheduleRequest: appending, has_child_req: {}", maybe_child_req.is_some());

                let 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: version.clone(),
                    version_max_excluding: Version::new(version.0 + 1),
                });

                // If we have a child_req, use Schedule event; otherwise just persist the history event
                if let Some(child_req) = maybe_child_req {
                    let non_blocking_event = CacheableDbEvent::Schedule {
                        request: append_req,
                        version: version.clone(),
                        child_req,
                        backtrace,
                    };
                    db_connection
                        .append_non_blocking(non_blocking_event, called_at)
                        .await?;
                } else {
                    // Error case: only persist the history event, no child creation
                    let non_blocking_event = CacheableDbEvent::ScheduleError {
                        request: append_req,
                        version: version.clone(),
                        backtrace,
                    };
                    db_connection
                        .append_non_blocking(non_blocking_event, called_at)
                        .await?;
                }

                Ok(history_event)
            }

            EventCallNonBlocking::Stub(Stub {
                intent,
                params,
                wasm_backtrace,
            }) => {
                // `WorkflowDbConnection::get_stub_create_request` already called when creating `intent`.
                // The response write cannot be cached: we need its conflict result
                // immediately, so it flushes the non-blocking cache first.
                // Idempotently attempt to write to target_execution_id.
                // The idempotent write is needed to avoid race with stub requests originating from remote systems.
                debug!(target_execution_id = %params.target_execution_id, "StubRequest: first write");
                match intent {
                    StubIntent::Err(err) => {
                        let event = HistoryEvent::Stub {
                            target_execution_id: params.target_execution_id,
                            retval_hash: params.retval_hash,
                            result: Err(err.into()),
                        };
                        let history_event = (event.clone(), version.clone());
                        let history_event_req = AppendRequest {
                            created_at: called_at,
                            event: ExecutionRequest::HistoryEvent { event },
                        };
                        db_connection
                            .append_batch(
                                version.clone(),
                                called_at,
                                vec![history_event_req],
                                db_connection.execution_id().clone(),
                                wasm_backtrace,
                                &self.locked_event.component_id,
                            )
                            .await?;
                        Ok(history_event)
                    }
                    StubIntent::StubTypeChecked(retval_intent) => {
                        let stub_finished_version = Version::new(1); // Stub activities have no execution log except Created event.
                        let (parent_id, join_set_id) = params.target_execution_id.split_to_parts();
                        let finished_req = AppendRequest {
                            created_at: called_at,
                            event: ExecutionRequest::Finished {
                                retval: retval_intent.clone(),
                                http_client_traces: None,
                            },
                        };
                        // Display-only backtrace keyed to the parent's stub-event version (`version`), where the HistoryEvent::Stub below lands.
                        let stub_backtrace =
                            wasm_backtrace.clone().map(|wasm_backtrace| BacktraceInfo {
                                execution_id: db_connection.execution_id().clone(),
                                component_id: self.locked_event.component_id.clone(),
                                version_min_including: version.clone(),
                                version_max_excluding: Version::new(version.0 + 1),
                                wasm_backtrace,
                            });
                        let result = match db_connection
                            .upsert_stub_response(
                                params.target_execution_id.clone(),
                                stub_finished_version.clone(),
                                finished_req,
                                AppendResponseToExecution {
                                    parent_execution_id: parent_id,
                                    created_at: called_at,
                                    join_set_id,
                                    child_execution_id: params.target_execution_id.clone(),
                                    finished_version: stub_finished_version,
                                    result: retval_intent.clone(),
                                },
                                called_at,
                                stub_backtrace,
                            )
                            .await
                        {
                            Ok(()) => Ok(()),
                            Err(UpsertStubOrReplayInterrupt::StubConflict) => {
                                info!(target_execution_id = %params.target_execution_id,
                                    "Got conflict while upserting stub response"
                                );
                                Err(StubError::Conflict)
                            }
                            Err(UpsertStubOrReplayInterrupt::ReplayInterrupt) => {
                                debug!(target_execution_id = %params.target_execution_id, "StubRequest: upsert interrupting replay");
                                return Err(DbErrorWriteOrReplayInterrupt::ReplayInterrupt);
                            }
                            Err(UpsertStubOrReplayInterrupt::DbError(db_err)) => {
                                return Err(DbErrorWriteOrReplayInterrupt::DbError(db_err));
                            }
                        };
                        debug!(target_execution_id = %params.target_execution_id, "Executed upsert_stub_response: {result:?}");
                        // Second write tx: Append the HistoryEvent with target_result.
                        let event = HistoryEvent::Stub {
                            target_execution_id: params.target_execution_id.clone(),
                            retval_hash: params.retval_hash.clone(),
                            result,
                        };
                        let history_event = (event.clone(), version.clone());
                        let history_event_req = AppendRequest {
                            created_at: called_at,
                            event: ExecutionRequest::HistoryEvent { event },
                        };

                        db_connection
                            .append_batch(
                                version.clone(),
                                called_at,
                                vec![history_event_req],
                                db_connection.execution_id().clone(),
                                wasm_backtrace,
                                &self.locked_event.component_id,
                            )
                            .await?;
                        Ok(history_event)
                    }
                }
            }

            EventCallNonBlocking::JoinNextTry(JoinNextTry {
                join_set_id,
                wasm_backtrace,
            }) => {
                // Determine the outcome: Found, Pending, or AllProcessed
                let outcome = if self.has_unprocessed_response_for_join_set(&join_set_id) {
                    JoinNextTryOutcome::Found
                } else if self
                    .join_set_open_tracker
                    .open_join_sets()
                    .get(&join_set_id)
                    .is_some_and(|requests| !requests.is_empty())
                {
                    JoinNextTryOutcome::Pending
                } else {
                    JoinNextTryOutcome::AllProcessed
                };

                debug!(%join_set_id, %outcome, "JoinNextTry");
                let event = HistoryEvent::JoinNextTry {
                    join_set_id,
                    outcome,
                };
                let history_event = (event.clone(), version.clone());
                let request = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                // TODO: append_non_blocking
                db_connection
                    .append_non_blocking(
                        CacheableDbEvent::JoinNextTry {
                            request,
                            version: 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: version.clone(),
                                version_max_excluding: Version::new(version.0 + 1),
                            }),
                        },
                        called_at,
                    )
                    .await?;
                Ok(history_event)
            }
        }
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_to_db_blocking(
        &mut self,
        event_call: EventCallBlocking,
        version: Version,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
        lock_expires_at: DateTime<Utc>,
    ) -> Result<AppendedBlockingEvents, DbErrorWrite> {
        trace!("append_to_db_blocking {}", version);
        match event_call {
            EventCallBlocking::JoinNext(JoinNext {
                join_set_id,
                wasm_backtrace,
            }) => {
                debug!(%join_set_id, "JoinNext(closing:false): 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: false,
                        }
                    } else {
                        HistoryEvent::JoinNextTooMany {
                            join_set_id,
                            requested_ffqn: None,
                        }
                    };
                let history_events = vec![(event.clone(), version.clone())];
                let join_next = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                db_connection
                    .append_blocking(
                        version.clone(),
                        db_connection.execution_id().clone(),
                        join_next,
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;
                Ok(AppendedBlockingEvents {
                    history_events,
                    known_response: None,
                })
            }

            EventCallBlocking::JoinSetClose(JoinSetClose {
                join_set_id,
                cancellations,
                wasm_backtrace,
            }) => {
                debug!(
                    %join_set_id,
                    "JoinSetClose: appending JoinNext(closing:true) with cancellations: {}", cancellations.is_some()
                );
                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: true,
                        }
                    } else {
                        HistoryEvent::JoinNextTooMany {
                            join_set_id,
                            requested_ffqn: None,
                        }
                    };
                let history_events = vec![(event.clone(), version.clone())];
                let join_next = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                db_connection
                    .append_join_set_close(
                        version.clone(),
                        &self.cancel_registry,
                        db_connection.execution_id().clone(),
                        join_next,
                        cancellations,
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;
                Ok(AppendedBlockingEvents {
                    history_events,
                    known_response: None,
                })
            }

            EventCallBlocking::JoinNextRequestingFfqn(JoinNextRequestingFfqn {
                join_set_id,
                requested_ffqn,
                wasm_backtrace,
            }) => {
                debug!(%join_set_id, "EventCallBlocking::JoinNextRequestingFfqn: 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(), version.clone())];
                let append_request = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };

                db_connection
                    .append_blocking(
                        version.clone(),
                        db_connection.execution_id().clone(),
                        append_request,
                        wasm_backtrace,
                        &self.locked_event.component_id,
                    )
                    .await?;

                Ok(AppendedBlockingEvents {
                    history_events,
                    known_response: None,
                })
            }

            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 = version.clone();
                history_events.push((event.clone(), version.clone()));
                let join_set = AppendRequest {
                    event: ExecutionRequest::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(),
                        result: Ok(()),
                    },
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let child_exec_req = AppendRequest {
                    event: ExecutionRequest::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: ExecutionRequest::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,
                    deployment_id: self.deployment_id,
                    scheduled_by: None,
                    paused: false,
                };

                db_connection
                    .append_batch_create_new_execution(
                        history_events
                            .first()
                            .expect("direct call has three events")
                            .1
                            .clone(),
                        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(AppendedBlockingEvents {
                    history_events,
                    known_response: None,
                })
            }

            EventCallBlocking::OneOffDelayRequest(OneOffDelayRequest {
                join_set_id,
                delay_id,
                schedule_at,
                expires_at_if_new,
                wasm_backtrace,
            }) => {
                // An already-due delay (e.g. `sleep(now)`) is fulfilled in the same
                // transaction, so the workflow resumes without waiting for the
                // expired-timers watcher to append the delay response.
                let already_due = expires_at_if_new <= called_at;
                debug!(%delay_id, %join_set_id, already_due, "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 = version.clone();
                history_events.push((event.clone(), version.clone()));
                let join_set = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                let event = HistoryEvent::JoinSetRequest {
                    join_set_id: join_set_id.clone(),
                    request: JoinSetRequest::DelayRequest {
                        delay_id: delay_id.clone(),
                        expires_at: expires_at_if_new,
                        schedule_at,
                        paused: false,
                    },
                };
                version = version.increment();
                history_events.push((event.clone(), version.clone()));
                let delay_req = AppendRequest {
                    created_at: called_at,
                    event: ExecutionRequest::HistoryEvent { event },
                };
                let event = HistoryEvent::JoinNext {
                    join_set_id: join_set_id.clone(),
                    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: ExecutionRequest::HistoryEvent { event },
                };

                let batch_version = history_events
                    .first()
                    .expect("blocking delay has three events")
                    .1
                    .clone();
                let batch = vec![join_set, delay_req, join_next];
                if already_due {
                    db_connection
                        .append_batch_with_delay_response(
                            batch_version,
                            called_at,
                            batch,
                            db_connection.execution_id().clone(),
                            join_set_id,
                            delay_id,
                            wasm_backtrace,
                            &self.locked_event.component_id,
                        )
                        .await?;
                } else {
                    db_connection
                        .append_batch(
                            batch_version,
                            called_at,
                            batch,
                            db_connection.execution_id().clone(),
                            wasm_backtrace,
                            &self.locked_event.component_id,
                        )
                        .await?;
                }

                let known_response = (already_due
                    && matches!(
                        self.join_next_blocking_strategy,
                        JoinNextBlockingStrategy::Await { .. }
                    ))
                .then_some(ChildReturnValue::OneOffDelay {
                    scheduled_at: expires_at_if_new,
                    result: Ok(()),
                });

                Ok(AppendedBlockingEvents {
                    history_events,
                    known_response,
                })
            }
        }
    }

    #[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
            .event
            .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"
            ),
        }
    }

    /// Returns `Ok(Some(json))` for Ok values, `Err(Some(json))` for Err values,
    /// or `Ok(None)`/`Err(None)` for unit types.
    pub(crate) fn get_processed_response_json(
        &self,
        child_execution_id: &ExecutionIdDerived,
    ) -> Result<
        Result<Option<String>, Option<String>>,
        latest::obelisk::workflow::workflow_support::GetResultJsonError,
    > {
        use latest::obelisk::workflow::workflow_support::GetResultJsonError;

        // Check if it was awaited (exists in processed responses index)
        let response_idx = self
            .index_child_exe_to_processed_response_idx
            .get(child_execution_id)
            .ok_or(GetResultJsonError::NotFoundInProcessedResponses)?;

        match &self
            .responses
            .get(*response_idx)
            .as_ref()
            .expect("`index_child_exe_to_processed_response_idx` must point to a response")
            .0
            .event
            .event
            .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");
                let ret_type = 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");

                // Convert the result to WastVal and then serialize to JSON
                let wast_val_res = result.clone().into_wast_val_res(|| ret_type);
                match wast_val_res {
                    Ok(inner) => {
                        let json = inner.map(|v| {
                            serde_json::to_string(&*v).expect("WastVal must be JSON serializable")
                        });
                        Ok(Ok(json))
                    }
                    Err(inner) => {
                        let json = inner.map(|v| {
                            serde_json::to_string(&*v).expect("WastVal must be JSON serializable")
                        });
                        Ok(Err(json))
                    }
                }
            }
            JoinSetResponse::DelayFinished { .. } => unreachable!(
                "`index_child_exe_to_processed_response_idx` must point to a ChildExecutionFinished"
            ),
        }
    }

    /// Platform-level failure kind of a *processed* child response, or `None` for
    /// an ok / business-err result. Same precondition and error channel as
    /// `get_processed_response_json`.
    pub(crate) fn get_processed_response_failure_kind(
        &self,
        child_execution_id: &ExecutionIdDerived,
    ) -> Result<
        Option<concepts::ExecutionFailureKind>,
        latest::obelisk::workflow::workflow_support::GetResultJsonError,
    > {
        use latest::obelisk::workflow::workflow_support::GetResultJsonError;

        let response_idx = self
            .index_child_exe_to_processed_response_idx
            .get(child_execution_id)
            .ok_or(GetResultJsonError::NotFoundInProcessedResponses)?;

        match &self
            .responses
            .get(*response_idx)
            .as_ref()
            .expect("`index_child_exe_to_processed_response_idx` must point to a response")
            .0
            .event
            .event
            .event
        {
            JoinSetResponse::ChildExecutionFinished { result, .. } => match result {
                SupportedFunctionReturnValue::ExecutionFailure(failure) => Ok(Some(failure.kind)),
                SupportedFunctionReturnValue::Ok(_) | SupportedFunctionReturnValue::Err(_) => {
                    Ok(None)
                }
            },
            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_ne!(kind, JoinSetKind::Named);
        (self.join_set_count(kind) + 1).to_string()
    }

    pub(crate) 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::JoinNextTry { .. }
                        | 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::JoinNextTry { .. }
                        | 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 offset = self
            .event_history
            .iter()
            .filter(|(event, processing_status, _version)| {
                // Applying `SubmitDelay` immediately marks the event 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();
        let offset = u64::try_from(offset).expect("too many delays in a join set");
        let new_delay_id = DelayId::new(execution_id, join_set_id).get_incremented_by(offset);

        // backcompat: 0.40.0 histories can contain delay indices starting at 0.
        // Reuse the persisted id only when it follows the old or current sequence.
        if let Some(found_delay_id) =
            self.event_history
                .iter()
                .find_map(
                    |(event, processing_status, _version)| match (event, processing_status) {
                        (
                            HistoryEvent::JoinSetRequest {
                                join_set_id: found_join_set_id,
                                request: JoinSetRequest::DelayRequest { delay_id, .. },
                            },
                            Unprocessed,
                        ) if join_set_id == found_join_set_id => Some(delay_id),
                        _ => None,
                    },
                )
        {
            // return the found DelayId if it is 0-based.
            if found_delay_id.index() == offset {
                return found_delay_id.clone();
            }
        }
        new_delay_id
    }
}

#[derive(Debug, PartialEq, Eq)]
enum AwaitNextExtensionError {
    FunctionMismatch {
        specified_function: FunctionFqn,
        actual_function: Option<FunctionFqn>, // None on delay
        actual_id: JoinSetResponseId,
    },
    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 {
                    JoinSetResponseId::ChildExecutionId(id) => {
                        (ValKey::new_snake("execution_id"), id.to_string())
                    }
                    JoinSetResponseId::DelayId(id) => {
                        (ValKey::new_snake("delay_id"), id.to_string())
                    }
                };
                WastVal::Variant(
                    ValKey::new_snake("function_mismatch"),
                    Some(Box::new(WastVal::Record(indexmap! {
                        ValKey::new_snake("specified_function") => ffqn_into_wast_val(specified),
                        ValKey::new_snake("actual_function") =>  WastVal::Option(
                            actual.as_ref().map(|actual| Box::from(ffqn_into_wast_val(actual)))),
                        ValKey::new_snake("actual_id") =>
                            WastVal::Variant(actual_id_field_name,
                            Some(Box::new(
                            WastVal::Record(indexmap!{
                                ValKey::new_snake("id") => WastVal::String(actual_str_value)
                            }))
                        ))
                    }))),
                )
            }
            AwaitNextExtensionError::AllProcessed => {
                WastVal::Variant(ValKey::new_snake("all_processed"), 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)]
struct EventCall {
    version: Version,
    version_range: EventCallVersionRange,
    kind: EventCallKind,
}

pub(crate) struct EventCallCursor {
    next_version: Version,
    // Starts with whole replay history, pops front upon consuming each event.
    replay_versions: VecDeque<Version>,
}

impl EventCallCursor {
    pub(crate) fn new(next_version: Version, event_history: &[(HistoryEvent, Version)]) -> Self {
        Self {
            next_version,
            replay_versions: event_history
                .iter()
                .map(|(_, version)| version.clone())
                .collect(),
        }
    }

    pub(crate) fn version(&self) -> &Version {
        &self.next_version
    }

    /// True while there are still persisted events to replay before the cursor reaches the
    /// live tip. Once exhausted, subsequent calls produce brand-new events.
    pub(crate) fn is_replaying_persisted(&self) -> bool {
        !self.replay_versions.is_empty()
    }

    fn next(&mut self, kind: EventCallKind) -> EventCall {
        let event_count =
            u32::try_from(kind.as_keys().len()).expect("an EventCall has at most three events");
        let version = if let Some(version) = self.replay_versions.pop_front() {
            for _ in 1..event_count {
                self.replay_versions.pop_front();
            }
            version
        } else {
            let version = self.next_version.clone();
            self.next_version = Version::new(self.next_version.0 + event_count);
            version
        };
        let version_range = EventCallVersionRange {
            min_including: version.clone(),
            max_excluding: Version::new(version.0 + event_count),
        };
        EventCall::new(version, version_range, kind)
    }
}

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

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

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

impl EventCall {
    fn new(version: Version, version_range: EventCallVersionRange, kind: EventCallKind) -> Self {
        Self {
            version,
            version_range,
            kind,
        }
    }

    fn wasm_backtrace(&self) -> Option<&storage::WasmBacktrace> {
        match &self.kind {
            EventCallKind::Blocking(event) => match event {
                EventCallBlocking::JoinNextRequestingFfqn(event) => event.wasm_backtrace.as_ref(),
                EventCallBlocking::JoinNext(event) => event.wasm_backtrace.as_ref(),
                EventCallBlocking::JoinSetClose(event) => event.wasm_backtrace.as_ref(),
                EventCallBlocking::OneOffChildExecutionRequest(event) => {
                    event.wasm_backtrace.as_ref()
                }
                EventCallBlocking::OneOffDelayRequest(event) => event.wasm_backtrace.as_ref(),
            },
            EventCallKind::NonBlocking(event) => match event {
                EventCallNonBlocking::JoinSetCreate(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::SubmitChildExecution(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::SubmitDelay(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::JoinNextTry(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::Schedule(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::Stub(event) => event.wasm_backtrace.as_ref(),
                EventCallNonBlocking::Persist(event) => event.wasm_backtrace.as_ref(),
            },
        }
    }

    fn as_keys(&self) -> Vec<DeterministicKey> {
        self.kind.as_keys()
    }
}

#[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,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<JoinSetId, ApplyError> {
        assert_ne!(self.join_set_id.kind, JoinSetKind::OneOff);
        let join_set_id = self.join_set_id.clone();
        let value = event_history
            .apply_event_call(
                EventCallKind::NonBlocking(EventCallNonBlocking::JoinSetCreate(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value,
            ChildReturnValue::JoinSetCreate(join_set_id) => join_set_id);
        assert_eq!(join_set_id, value);
        event_history
            .join_set_open_tracker
            .create_join_set(join_set_id)
            .expect("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) join_set_id: JoinSetId,
    pub(crate) child_execution_id: ExecutionIdDerived,
    pub(crate) intent: SubmitChildIntent,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl SubmitChildExecution {
    /// Apply the submit child execution event and return the stored result.
    /// Returns `Ok(Ok(execution_id_val))` on success, or `Ok(Err(error))` on stored error.
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Result<(), ChildExecutionRequestError>, 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 target_ffqn = self.target_ffqn.clone();
        // Get component_type from intent if Ok
        let component_type = match &self.intent {
            SubmitChildIntent::Ok {
                fn_component_id, ..
            } => Some(fn_component_id.component_type),
            SubmitChildIntent::Err(_) => None,
        };
        let value = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::SubmitChildExecution(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        let result = assert_matches!(value, ChildReturnValue::SubmitChild(result) => result);

        // Only update the index if the result was Ok (child was actually created)
        if result.is_ok()
            && let Some(component_type) = component_type
        {
            event_history
                .join_set_open_tracker
                .insert_child(
                    &join_set_id,
                    child_execution_id,
                    component_type,
                    target_ffqn,
                )
                .map_err(|err| {
                    WorkflowFunctionError::ConstraintViolation(
                        join_set_open_tracker_error_to_constraint(&err),
                    )
                })?;
        }

        Ok(result)
    }
}

#[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,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        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(
                EventCallKind::NonBlocking(EventCallNonBlocking::SubmitDelay(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        assert_matches!(value, ChildReturnValue::SubmitDelay);
        event_history
            .join_set_open_tracker
            .insert_delay(&join_set_id, delay_id.clone())
            .map_err(|err| {
                WorkflowFunctionError::ConstraintViolation(
                    join_set_open_tracker_error_to_constraint(&err),
                )
            })?;
        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) intent: ScheduleIntent,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl Schedule {
    /// Apply the schedule event and return the stored result.
    /// Returns `Ok(Ok(execution_id_val))` on success, or `Ok(Err(error))` on stored error.
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Result<(), ScheduleRequestError>, WorkflowFunctionError> {
        let value = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Schedule(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        let result = assert_matches!(value, ChildReturnValue::Schedule(result) => result);
        Ok(result)
    }
}

#[derive(derive_more::Debug, Clone, PartialEq, Eq)]
pub(crate) struct StubParams {
    pub(crate) target_execution_id: ExecutionIdDerived,
    pub(crate) retval_hash: StubRetValHash,
}

// Current database and fn registry depentent intent
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum StubIntent {
    Err(StubIntentErr),
    StubTypeChecked(SupportedFunctionReturnValue), // can result in `Ok(())` or `StubError::Conflict`
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum StubIntentErr {
    ExecutionNotFound,      // results in `StubError::ExecutionNotFound`
    TypeCheckError(String), // results in `StubError::TypeCheckError`
}
impl From<StubIntentErr> for StubError {
    fn from(value: StubIntentErr) -> StubError {
        match value {
            StubIntentErr::ExecutionNotFound => StubError::ExecutionNotFound,
            StubIntentErr::TypeCheckError(reason) => StubError::TypeCheckError(reason),
        }
    }
}

/// Intent for `schedule-json` function. Captures `fn_registry` lookup result.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ScheduleIntent {
    Ok {
        fn_component_id: ComponentId,
        params: Params,
    },
    Err(ScheduleRequestError),
}

/// Intent for `submit-json` function. Captures `fn_registry` lookup result.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SubmitChildIntent {
    Ok {
        fn_component_id: ComponentId,
        params: Params,
    },
    Err(ChildExecutionRequestError),
}

// -stub: func(execution-id: execution-id, return-value: result<?>) -> result<_, stub-error>
#[derive(derive_more::Debug, Clone)]
pub(crate) struct Stub {
    pub(crate) intent: StubIntent,
    pub(crate) params: StubParams,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl Stub {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Result<(), StubError>, WorkflowFunctionError> {
        let value = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Stub(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;

        let result = assert_matches!(value, ChildReturnValue::Stub(result) => result);
        Ok(result)
    }
}

// -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,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        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(
                EventCallKind::Blocking(EventCallBlocking::JoinNextRequestingFfqn(self)),
                event_call_cursor,
                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)) => {
                // `-await-next` now returns `result<T, err>` directly (no id tuple);
                // read the id via `join-set.last-id`. Mirrors `-get`.
                let wast_val_res = WastVal::Result(Ok(Some(Box::new(wast_val_result))));
                event_history.record_last_response_id(
                    &join_set_id,
                    JoinSetResponseId::ChildExecutionId(child_execution_id.clone()),
                );
                event_history
                    .join_set_open_tracker
                    .remove_response(
                        &join_set_id,
                        &JoinSetResponseId::ChildExecutionId(child_execution_id),
                    )
                    .map_err(|err| {
                        WorkflowFunctionError::ConstraintViolation(
                            join_set_open_tracker_error_to_constraint(&err),
                        )
                    })?;
                wast_val_res
            }
            Err(await_ext_err) => {
                if let AwaitNextExtensionError::FunctionMismatch { actual_id, .. } = &await_ext_err
                {
                    event_history
                        .join_set_open_tracker
                        .remove_response(&join_set_id, actual_id)
                        .map_err(|err| {
                            WorkflowFunctionError::ConstraintViolation(
                                join_set_open_tracker_error_to_constraint(&err),
                            )
                        })?;
                } // all-processed does not change the join-set open-tracker
                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,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl JoinNext {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<
        Result<(types_execution::ResponseId, Result<(), ()>), 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_event_call(
                EventCallKind::Blocking(EventCallBlocking::JoinNext(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value,ChildReturnValue::JoinNext(value) => value);
        if let Ok((response_id, _)) = &value {
            event_history.record_last_response_id(&join_set_id, response_id.clone());
            event_history
                .join_set_open_tracker
                .remove_response(&join_set_id, response_id)
                .map_err(|err| {
                    WorkflowFunctionError::ConstraintViolation(
                        join_set_open_tracker_error_to_constraint(&err),
                    )
                })?;
        }
        let value = value
            .map(|(response_id, result)| (types_execution::ResponseId::from(response_id), result));
        Ok(value)
    }
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct JoinSetClose {
    pub(crate) join_set_id: JoinSetId,
    pub(crate) cancellations: Option<JoinSetCloseCancellations>,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}

#[derive(derive_more::Debug, Clone)]
pub(crate) struct JoinNextTry {
    pub(crate) join_set_id: JoinSetId,
    #[debug(skip)]
    pub(crate) wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl JoinNextTry {
    pub(crate) async fn apply(
        self,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<
        Result<(types_execution::ResponseId, Result<(), ()>), JoinNextTryError>,
        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_event_call(
                EventCallKind::NonBlocking(EventCallNonBlocking::JoinNextTry(self)),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        let value = assert_matches!(value, ChildReturnValue::JoinNextTry(value) => value);
        if let Ok((response_id, _)) = &value {
            event_history.record_last_response_id(&join_set_id, response_id.clone());
            event_history
                .join_set_open_tracker
                .remove_response(&join_set_id, response_id)
                .map_err(|err| {
                    WorkflowFunctionError::ConstraintViolation(
                        join_set_open_tracker_error_to_constraint(&err),
                    )
                })?;
        } // else it is JoinNextTryError
        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 {
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn apply(
        ffqn: FunctionFqn,
        fn_component_id: ComponentId,
        params: Params,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        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 = EventCallKind::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(event, event_call_cursor, db_connection, called_at)
            .await?;
        event_history
            .record_last_oneoff_id(JoinSetResponseId::ChildExecutionId(child_execution_id));
        let value = assert_matches!(value,
            ChildReturnValue::WastVal(wast_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 {
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn apply(
        schedule_at: HistoryEventScheduleAt,
        name: Option<String>,
        expires_at_if_new: DateTime<Utc>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<Result<DateTime<Utc>, ()>, WorkflowFunctionError> {
        let suffix = name.as_deref().unwrap_or("sleep");
        let join_set_id = event_history
            .next_join_set_one_off_named(suffix)
            .map_err(|err| WorkflowFunctionError::ImportedFunctionCallError {
                // Only `sleep-named-bt` passes the name
                ffqn: FunctionFqn::new_static("obelisk:workflow/workflow-support@6.0.0", "sleep"),
                reason: "invalid sleep join set name".into(),
                detail: Some(err.to_string()),
            })?;
        let delay_id = DelayId::new(db_connection.execution_id(), &join_set_id);

        let ChildReturnValue::OneOffDelay {
            scheduled_at,
            result,
        } = event_history
            .apply(
                EventCallKind::Blocking(EventCallBlocking::OneOffDelayRequest(
                    OneOffDelayRequest {
                        join_set_id,
                        delay_id: delay_id.clone(),
                        schedule_at,
                        expires_at_if_new,
                        wasm_backtrace,
                    },
                )),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?
        else {
            unreachable!()
        };
        event_history.record_last_oneoff_id(JoinSetResponseId::DelayId(delay_id));
        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 {
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn apply_string(
        value: &str,
        min_length: u64,
        max_length_exclusive: u64,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), WorkflowFunctionError> {
        let ret = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Persist(Persist {
                    value: Vec::from_iter(value.bytes()),
                    kind: PersistKind::RandomString {
                        min_length,
                        max_length_exclusive,
                    },
                    wasm_backtrace,
                })),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        assert_matches!(ret, ChildReturnValue::Persist);
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn apply_u64(
        value: u64,
        min: u64,
        max_inclusive: u64,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), WorkflowFunctionError> {
        let ret = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Persist(Persist {
                    value: Vec::from(storage::from_u64_to_bytes(value)),
                    kind: PersistKind::RandomU64 { min, max_inclusive },
                    wasm_backtrace,
                })),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        assert_matches!(ret, ChildReturnValue::Persist);
        Ok(())
    }

    pub(crate) async fn apply_execution_id(
        value: &ExecutionIdTopLevel,
        wasm_backtrace: Option<storage::WasmBacktrace>,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        db_connection: &mut dyn WorkflowDbConnection,
        called_at: DateTime<Utc>,
    ) -> Result<(), WorkflowFunctionError> {
        let ret = event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Persist(Persist {
                    value: Vec::from(value.ulid().0.to_be_bytes()),
                    kind: PersistKind::ExecutionId,
                    wasm_backtrace,
                })),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await?;
        assert_matches!(ret, ChildReturnValue::Persist);
        Ok(())
    }
}

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,
                wasm_backtrace: _,
            }) => JoinNextVariant::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: false,
            },
            EventCallBlocking::JoinSetClose(JoinSetClose {
                join_set_id,
                cancellations: _,
                wasm_backtrace: _,
            }) => JoinNextVariant::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: true,
            },
        }
    }
}

/// 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("JoinNextTry({join_set_id})")]
    JoinNextTry { join_set_id: JoinSetId },

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

    #[display("Stub({})", params.target_execution_id)]
    Stub {
        intent: StubIntent,
        params: StubParams,
    },
}

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

impl EventCallKind {
    fn as_keys(&self) -> Vec<DeterministicKey> {
        match self {
            EventCallKind::Blocking(inner) => inner.as_keys(),
            EventCallKind::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,
                wasm_backtrace: _,
            }) => vec![DeterministicKey::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: false,
            }],
            EventCallBlocking::JoinSetClose(JoinSetClose {
                join_set_id,
                cancellations: _,
                wasm_backtrace: _,
            }) => vec![DeterministicKey::JoinNext {
                join_set_id: join_set_id.clone(),
                closing: true,
            }],
        }
    }
}

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,
                intent,
                wasm_backtrace: _,
            }) => {
                // Extract params from intent for deterministic matching
                let params = match intent {
                    SubmitChildIntent::Ok { params, .. } => params.clone(),
                    SubmitChildIntent::Err(_) => Params::empty(),
                };
                DeterministicKey::ChildExecutionRequest {
                    join_set_id: join_set_id.clone(),
                    child_execution_id: child_execution_id.clone(),
                    target_ffqn: target_ffqn.clone(),
                    params,
                }
            }
            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 { intent, params, .. }) => DeterministicKey::Stub {
                intent: intent.clone(),
                params: params.clone(),
            },
            EventCallNonBlocking::JoinNextTry(JoinNextTry { join_set_id, .. }) => {
                DeterministicKey::JoinNextTry {
                    join_set_id: join_set_id.clone(),
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::super::event_history::{
        EventCallBlocking, EventCallCursor, EventCallKind, 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, WorkflowDbConnection,
    };
    use crate::workflow::deadline_tracker::DeadlineTrackerFactory;
    use crate::workflow::deadline_tracker::deadline_tracker_factory_test;
    use crate::workflow::event_history::{
        ApplyError, AwaitNextExtensionError, ChildReturnValue, JoinNextRequestingFfqn, JoinNextTry,
        JoinNextTryError, JoinSetCreate, OneOffDelayRequest, Schedule, ScheduleIntent, Stub,
        StubIntent, StubParams, SubmitChildIntent, SubmitDelay,
    };
    use assert_matches::assert_matches;
    use chrono::{DateTime, Utc};
    use concepts::prefixed_ulid::{DEPLOYMENT_ID_DUMMY, ExecutionIdDerived, ExecutorId, RunId};
    use concepts::storage::{
        AppendRequest, CreateRequest, DbConnectionTest, ExecutionRequest, HistoryEventScheduleAt,
        Locked, StubRetVal,
    };
    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_common::JoinSetResponseId;
    use db_tests::Database;
    use rstest::rstest;
    use std::sync::Arc;
    use std::time::Duration;
    use test_db_macro::expand_enum_database;
    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");

    #[tokio::test]
    async fn already_due_one_off_delays_are_resolved_before_cache_flush() {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();
        let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;
        let initial_version = db_connection.get(&execution_id).await.unwrap().next_version;

        let (mut event_history, mut event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1),
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: 100,
                    subscription_interruption: None,
                },
                TestingFnRegistry::new_from_components(vec![]),
            )
            .await;

        for name in ["first", "second"] {
            let result = OneOffDelayRequest::apply(
                HistoryEventScheduleAt::Now,
                Some(name.to_owned()),
                sim_clock.now(),
                None,
                &mut event_history,
                &mut event_call_cursor,
                &mut *caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();
            assert_eq!(Ok(sim_clock.now()), result);
        }

        let unflushed_log = db_connection.get(&execution_id).await.unwrap();
        assert_eq!(initial_version, unflushed_log.next_version);
        assert!(unflushed_log.responses.is_empty());

        caching_db_connection
            .flush_non_blocking_event_cache(sim_clock.now())
            .await
            .unwrap();

        let flushed_log = db_connection.get(&execution_id).await.unwrap();
        assert_eq!(
            Version::new(initial_version.0 + 6),
            flushed_log.next_version
        );
        assert_eq!(2, flushed_log.responses.len());

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

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

        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                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,
                &mut event_call_cursor,
                join_set_id.clone(),
                sim_clock.now()
            )
            .await
            .unwrap_err(),
            ApplyError::InterruptDbUpdated,
            "should have ended with an interrupt"
        );
        let finished_version = finish_child_execution(
            db_connection.as_ref(),
            child_execution_id.clone(),
            sim_clock.now(),
            SUPPORTED_RETURN_VALUE_OK_EMPTY,
        )
        .await;
        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,
                        result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");
        let (mut event_history, mut event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id,
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                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,
            &mut event_call_cursor,
            join_set_id,
            sim_clock.now(),
        )
        .await
        .expect("response was appended, should finish successfuly");

        event_history
            .finalize(
                &mut event_call_cursor,
                &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, subscription_interruption: None }, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 10, subscription_interruption: None })]
        join_next_blocking_strategy: JoinNextBlockingStrategy,
    ) {
        const CHILD_RESP: SupportedFunctionReturnValue =
            SupportedFunctionReturnValue::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_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();
        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                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,
            &mut event_call_cursor,
            join_set_id.clone(),
            MOCK_FFQN,
            child_execution_id.clone(),
            sim_clock.now(),
        )
        .await;

        // append child response before issuing join_next
        caching_db_connection
            .flush_non_blocking_event_cache(sim_clock.now())
            .await
            .unwrap();
        let finished_version = finish_child_execution(
            db_connection.as_ref(),
            child_execution_id.clone(),
            sim_clock.now(),
            CHILD_RESP,
        )
        .await;
        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,
                        result: CHILD_RESP,
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");

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

        apply_create_join_set_start_async(
            &mut *caching_db_connection,
            &mut event_history,
            &mut event_call_cursor,
            join_set_id.clone(),
            MOCK_FFQN,
            child_execution_id.clone(),
            sim_clock.now(),
        )
        .await;
        // issue BlockingChildJoinNext
        let res = event_history
            .apply(
                EventCallKind::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                    JoinNextRequestingFfqn {
                        join_set_id,
                        wasm_backtrace: None,
                        requested_ffqn: MOCK_FFQN,
                    },
                )),
                &mut event_call_cursor,
                &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 event_call_cursor,
                &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(Some(WastValWithType {
                r#type: TypeWrapper::U8,
                value: WastVal::U8(1),
            }));
        const KID_B_RET: SupportedFunctionReturnValue =
            SupportedFunctionReturnValue::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_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();

        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                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,
                &mut event_call_cursor,
                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
        let first_child_execution_id = if submits_and_awaits_in_correct_order {
            child_execution_id_a.clone()
        } else {
            child_execution_id_b.clone()
        };
        let first_finished_version = finish_child_execution(
            db_connection.as_ref(),
            first_child_execution_id.clone(),
            sim_clock.now(),
            KID_A_RET,
        )
        .await;
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: first_child_execution_id,
                        finished_version: first_finished_version,
                        result: KID_A_RET, // won't matter on mismatch
                    },
                },
            )
            .await
            .unwrap();
        let second_child_execution_id = if submits_and_awaits_in_correct_order {
            child_execution_id_b.clone()
        } else {
            child_execution_id_a.clone()
        };
        let second_finished_version = finish_child_execution(
            db_connection.as_ref(),
            second_child_execution_id.clone(),
            sim_clock.now(),
            KID_B_RET,
        )
        .await;
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::ChildExecutionFinished {
                        child_execution_id: second_child_execution_id,
                        finished_version: second_finished_version,
                        result: KID_B_RET, // won't matter on mismatch
                    },
                },
            )
            .await
            .unwrap();

        info!("Second run");

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

        let res = apply_create_join_set_two_start_asyncs_await_next_a(
            &mut *caching_db_connection,
            &mut event_history,
            &mut event_call_cursor,
            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: JoinSetResponseId::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(
                    EventCallKind::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                        JoinNextRequestingFfqn {
                            join_set_id,
                            wasm_backtrace: None,
                            requested_ffqn: submit_ffqn_2.clone(),
                        },
                    )),
                    &mut event_call_cursor,
                    &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 event_call_cursor,
                &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_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Interrupt, // does not matter, there are no blocking events
                fn_registry,
            )
            .await;

        event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::Schedule(Schedule {
                    schedule_at: HistoryEventScheduleAt::Now,
                    scheduled_at_if_new: sim_clock.now(),
                    execution_id: ExecutionId::generate(),
                    ffqn: MOCK_FFQN,
                    intent: ScheduleIntent::Ok {
                        fn_component_id: ComponentId::dummy_activity(),
                        params: Params::empty(),
                    },
                    wasm_backtrace: None,
                })),
                &mut event_call_cursor,
                &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(
                EventCallKind::NonBlocking(EventCallNonBlocking::JoinSetCreate(JoinSetCreate {
                    join_set_id: join_set_id.clone(),
                    wasm_backtrace: None,
                })),
                &mut event_call_cursor,
                &mut *caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();

        event_history
            .finalize(
                &mut event_call_cursor,
                &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_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        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 event_call_cursor, mut caching_db_connection) =
                load_event_history(
                    db_pool.connection_test().await.unwrap(),
                    execution_id.clone(),
                    sim_clock.now(),
                    Duration::from_secs(1), // execution deadline
                    deadline_tracker_factory_test(&sim_clock),
                    JoinNextBlockingStrategy::Await {
                        non_blocking_event_batching: 0,
                        subscription_interruption: None,
                    },
                    fn_registry.clone(),
                )
                .await;

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

            event_history
                .apply(
                    EventCallKind::NonBlocking(EventCallNonBlocking::Stub(Stub {
                        intent: StubIntent::StubTypeChecked(SUPPORTED_RETURN_VALUE_OK_EMPTY),
                        params: StubParams {
                            target_execution_id: child_execution_id.clone(),
                            retval_hash: StubRetVal::Typed(SUPPORTED_RETURN_VALUE_OK_EMPTY).hash(),
                        },
                        wasm_backtrace: None,
                    })),
                    &mut event_call_cursor,
                    &mut *caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();

            let child_return_value = event_history
                .apply(
                    EventCallKind::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                        JoinNextRequestingFfqn {
                            join_set_id: join_set_id.clone(),
                            wasm_backtrace: None,
                            requested_ffqn: MOCK_FFQN,
                        },
                    )),
                    &mut event_call_cursor,
                    &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 event_call_cursor,
                    &mut *caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();
        }

        db_close.close().await;
    }

    #[expand_enum_database] // DB Conflict must be tested for each database
    #[rstest]
    #[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(db: Database) {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, db_close) = db.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();

        let execution_id = create_execution(db_connection.as_ref(), &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 event_call_cursor, mut caching_db_connection) =
                load_event_history(
                    db_pool.connection_test().await.unwrap(),
                    execution_id.clone(),
                    sim_clock.now(),
                    Duration::from_secs(1),
                    deadline_tracker_factory_test(&sim_clock),
                    JoinNextBlockingStrategy::Await {
                        non_blocking_event_batching: 0,
                        subscription_interruption: None,
                    },
                    fn_registry.clone(),
                )
                .await;
            apply_create_join_set_start_async(
                &mut *caching_db_connection,
                &mut event_history,
                &mut event_call_cursor,
                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(
                        EventCallKind::NonBlocking(EventCallNonBlocking::Stub(Stub {
                            intent: StubIntent::StubTypeChecked(SUPPORTED_RETURN_VALUE_OK_EMPTY),
                            params: StubParams {
                                target_execution_id: target_activity_stub.clone(),
                                retval_hash: StubRetVal::Typed(SUPPORTED_RETURN_VALUE_OK_EMPTY)
                                    .hash(),
                            },
                            wasm_backtrace: None,
                        })),
                        &mut event_call_cursor,
                        &mut *caching_db_connection,
                        sim_clock.now(),
                    )
                    .await
                    .unwrap();
            }

            event_history
                .finalize(
                    &mut event_call_cursor,
                    &mut *caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();
        }
        drop(execution_id);
        // Second execution
        {
            let execution_id = create_execution(db_connection.as_ref(), &sim_clock).await;
            let (mut event_history, mut event_call_cursor, mut caching_db_connection) =
                load_event_history(
                    db_pool.connection_test().await.unwrap(),
                    execution_id.clone(),
                    sim_clock.now(),
                    Duration::from_secs(1),
                    deadline_tracker_factory_test(&sim_clock),
                    JoinNextBlockingStrategy::Await {
                        non_blocking_event_batching: 0,
                        subscription_interruption: None,
                    },
                    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(
                        EventCallKind::NonBlocking(EventCallNonBlocking::Stub(Stub {
                            intent: StubIntent::StubTypeChecked(SUPPORTED_RETURN_VALUE_OK_EMPTY),
                            params: StubParams {
                                target_execution_id: target_activity_stub.clone(),
                                retval_hash: StubRetVal::Typed(SUPPORTED_RETURN_VALUE_OK_EMPTY)
                                    .hash(),
                            },
                            wasm_backtrace: None,
                        })),
                        &mut event_call_cursor,
                        &mut *caching_db_connection,
                        sim_clock.now(),
                    )
                    .await
                    .unwrap();
            }

            event_history
                .finalize(
                    &mut event_call_cursor,
                    &mut *caching_db_connection,
                    sim_clock.now(),
                )
                .await
                .unwrap();
        }
        drop(db_connection);
        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, subscription_interruption: None }, JoinNextBlockingStrategy::Await { non_blocking_event_batching: 100, subscription_interruption: None })]
        second_run_strategy: JoinNextBlockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();

        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                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,
                &mut event_call_cursor,
                join_set_id.clone(),
                sim_clock.now()
            )
            .await
            .unwrap_err(),
            ApplyError::InterruptDbUpdated,
            "should have ended with an interrupt"
        );
        let finished_version = finish_child_execution(
            db_connection.as_ref(),
            child_execution_id.clone(),
            sim_clock.now(),
            SUPPORTED_RETURN_VALUE_OK_EMPTY,
        )
        .await;
        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,
                        result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    },
                },
            )
            .await
            .unwrap();

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

        // finish the execution
        let err = event_history
            .finalize(
                &mut event_call_cursor,
                &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(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        execution_id
    }

    async fn finish_child_execution(
        db_connection: &dyn DbConnectionTest,
        child_execution_id: ExecutionIdDerived,
        created_at: DateTime<Utc>,
        result: SupportedFunctionReturnValue,
    ) -> Version {
        let child_execution_id = ExecutionId::Derived(child_execution_id);
        let child_log = db_connection.get(&child_execution_id).await.unwrap();
        let finished_version = child_log.next_version.clone();
        db_connection
            .append(
                child_execution_id,
                finished_version.clone(),
                AppendRequest {
                    created_at,
                    event: ExecutionRequest::Finished {
                        retval: result,
                        http_client_traces: None,
                    },
                },
            )
            .await
            .unwrap();
        finished_version
    }

    async fn load_event_history(
        db_connection: Box<dyn DbConnectionTest>,
        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, EventCallCursor, Box<dyn WorkflowDbConnection>) {
        let execution_deadline = now + lock_expires_at;
        let deadline_tracker = deadline_factory
            .create(
                execution_deadline,
                tokio::sync::watch::channel(false).1,
                tokio::sync::watch::channel(false).1,
            )
            .unwrap();

        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let version = exec_log.next_version.clone();
        let history_events: Vec<_> = exec_log.event_history().collect();
        let event_call_cursor = EventCallCursor::new(version, &history_events);
        let caching_db_connection = CachingDbConnection::new(
            db_connection,
            execution_id,
            CachingBuffer::new(join_next_blocking_strategy),
        );
        let cancel_registry = CancelRegistry::new();
        let event_history = EventHistory::new(
            DEPLOYMENT_ID_DUMMY,
            history_events,
            exec_log.responses,
            join_next_blocking_strategy,
            fn_registry,
            cancel_registry,
            deadline_tracker,
            Locked {
                component_id: ComponentId::dummy_activity(),
                executor_id: ExecutorId::generate(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                run_id: RunId::generate(),
                lock_expires_at: execution_deadline,
                retry_config: ComponentRetryConfig::ZERO,
            },
            None, // lock_extension
            None, // subscription_interruption
            info_span!("worker-test"),
            false, // replaying_unfinished_execution
            None,  // max_replay_captured_writes
            None,  // max_events_per_run
            None,  // response_refresh_interval
        );

        (
            event_history,
            event_call_cursor,
            Box::new(caching_db_connection),
        )
    }

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

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

    #[expect(clippy::too_many_arguments, clippy::result_large_err)]
    async fn apply_create_join_set_two_start_asyncs_await_next_a(
        db_connection: &mut dyn WorkflowDbConnection,
        event_history: &mut EventHistory,
        event_call_cursor: &mut EventCallCursor,
        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,
            event_call_cursor,
            join_set_id.clone(),
            ffqn_a.clone(),
            child_execution_id_a,
            called_at,
        )
        .await;
        event_history
            .apply(
                EventCallKind::NonBlocking(EventCallNonBlocking::SubmitChildExecution(
                    SubmitChildExecution {
                        target_ffqn: ffqn_b,
                        join_set_id: join_set_id.clone(),
                        child_execution_id: child_execution_id_b,
                        intent: SubmitChildIntent::Ok {
                            fn_component_id: ComponentId::dummy_activity(),
                            params: Params::empty(),
                        },
                        wasm_backtrace: None,
                    },
                )),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await
            .unwrap();
        event_history
            .apply_event_call(
                EventCallKind::Blocking(EventCallBlocking::JoinNextRequestingFfqn(
                    JoinNextRequestingFfqn {
                        join_set_id,
                        wasm_backtrace: None,
                        requested_ffqn: ffqn_a,
                    },
                )),
                event_call_cursor,
                db_connection,
                called_at,
            )
            .await
            .map(|res| match res {
                ChildReturnValue::JoinNextRequestingFfqn(res) => res,
                other => {
                    unreachable!(
                        "EventCallBlocking::JoinNextRequestingFfqn returns ChildReturnValue::JoinNextRequestingFfqn, got {other:?}"
                    )
                }
            })
    }

    #[tokio::test]
    async fn join_next_try_processes_response() {
        use concepts::prefixed_ulid::DelayId;

        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();

        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Interrupt,
                fn_registry.clone(),
            )
            .await;

        // Create a join set using JoinSetCreate::apply which updates the index
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::Named, StrVariant::Arc("test".into())).unwrap();
        JoinSetCreate {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        // Submit a delay using SubmitDelay::apply which updates the index
        let delay_id = DelayId::new(&execution_id, &join_set_id);
        let schedule_at = HistoryEventScheduleAt::In(Duration::from_millis(10));
        let expires_at = schedule_at.as_date_time(sim_clock.now()).unwrap();
        SubmitDelay {
            join_set_id: join_set_id.clone(),
            delay_id: delay_id.clone(),
            schedule_at,
            expires_at_if_new: expires_at,
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        // Try join_next_try before response arrives - should return Pending
        let result = JoinNextTry {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();
        assert_matches!(result, Err(JoinNextTryError::Pending));

        // Append delay response to DB
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::DelayFinished {
                        delay_id: delay_id.clone(),
                        result: Ok(()),
                    },
                },
            )
            .await
            .unwrap();

        // Reload event history to pick up the response
        let (mut event_history, mut event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1),
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Interrupt,
                fn_registry.clone(),
            )
            .await;

        // Replay the join set creation and delay submission (updates index)
        JoinSetCreate {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        SubmitDelay {
            join_set_id: join_set_id.clone(),
            delay_id: delay_id.clone(),
            schedule_at,
            expires_at_if_new: expires_at,
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        // Replay: First JoinNextTry returned Pending (replay should return same)
        let result = JoinNextTry {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();
        assert_matches!(
            result,
            Err(JoinNextTryError::Pending),
            "replay should return same result as original"
        );

        // Now call join_next_try again - should find the response
        let result = JoinNextTry {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();
        let (response_id, delay_result) = result.expect("should find response");
        // Check that response_id is a delay with the expected id
        use crate::workflow::host_exports::latest::obelisk::types::execution::ResponseId as WitResponseId;
        let WitResponseId::DelayId(wit_delay_id) = response_id else {
            panic!("expected DelayId, got {response_id:?}");
        };
        assert_eq!(delay_id.to_string(), wit_delay_id.id);
        assert!(delay_result.is_ok(), "delay should not be cancelled");

        // Verify response is marked as processed - should return AllProcessed
        let result = JoinNextTry {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();
        assert_matches!(
            result,
            Err(JoinNextTryError::AllProcessed),
            "response should be marked as processed"
        );

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

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

    #[tokio::test]
    async fn join_next_try_all_processed_should_be_persisted() {
        test_utils::set_up();
        let sim_clock = SimClock::new(DateTime::default());
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection_test().await.unwrap();

        // 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 event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1), // execution deadline
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Interrupt,
                fn_registry.clone(),
            )
            .await;

        // Create a join set
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::Named, StrVariant::Arc("test".into())).unwrap();
        JoinSetCreate {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        // Try join_next_try should return AllProcessed
        let result = JoinNextTry {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();
        assert_matches!(result, Err(JoinNextTryError::AllProcessed));

        // no new events should be appended.
        event_history
            .finalize(
                &mut event_call_cursor,
                &mut *caching_db_connection,
                sim_clock.now(),
            )
            .await
            .unwrap();

        // Replay without the buggy join-next-try should trigger nondeterminism error.
        let (mut event_history, mut event_call_cursor, mut caching_db_connection) =
            load_event_history(
                db_pool.connection_test().await.unwrap(),
                execution_id.clone(),
                sim_clock.now(),
                Duration::from_secs(1),
                deadline_tracker_factory_test(&sim_clock),
                JoinNextBlockingStrategy::Interrupt,
                fn_registry.clone(),
            )
            .await;
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::Named, StrVariant::Arc("test".into())).unwrap();
        JoinSetCreate {
            join_set_id: join_set_id.clone(),
            wasm_backtrace: None,
        }
        .apply(
            &mut event_history,
            &mut event_call_cursor,
            &mut *caching_db_connection,
            sim_clock.now(),
        )
        .await
        .unwrap();

        let err = event_history
            .finalize(
                &mut event_call_cursor,
                &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 2: event: JoinNextTry(n:test, all_processed)",
            reason
        );

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