obeli-sk-wasm-workers 0.41.5

use super::activity_ctx::{self, ActivityCtx};
use crate::activity::cancel_registry::CancelRegistry;
use crate::component_logger::{LogStrageConfig, log_activities};
use crate::envvar::EnvVar;
use crate::http_hooks::ConfigSectionHint;
use crate::std_output_stream::{StdOutputConfig, StdOutputConfigWithSender};
use crate::{RunnableComponent, WasmFileError};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use concepts::storage::http_client_trace::HttpClientTrace;
use concepts::storage::{LogInfoAppendRow, LogStreamType, Version};
use concepts::time::{ClockFn, Sleep, now_tokio_instant};
use concepts::{
    ComponentId, FunctionFqn, PackageIfcFns, Params, SupportedFunctionReturnValue, TrapKind,
};
use concepts::{FunctionMetadata, ResultParsingError};
use executor::worker::{FatalError, RunFinished, WorkerContext, WorkerResult, WorkerResultOk};
use executor::worker::{Worker, WorkerError};
use itertools::Itertools;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing::{debug, info, trace};
use utils::wasm_tools::ExIm;
use wasmtime::Store;
use wasmtime::component::{ComponentExportIndex, InstancePre, Type};
use wasmtime::{Engine, component::Val};

#[derive(Clone, Debug)]
pub struct ActivityConfig {
    pub component_id: ComponentId,
    pub forward_stdout: Option<StdOutputConfig>,
    pub forward_stderr: Option<StdOutputConfig>,
    pub env_vars: Arc<[EnvVar]>,
    pub fuel: Option<u64>,
    pub allowed_hosts: Arc<[crate::http_request_policy::AllowedHostConfig]>,
    pub global_http_config: crate::http_request_policy::GlobalHttpConfig,
    /// Resolves the component's declared secret names to values at execution-run
    /// policy-build time (scoped `RestrictedSecretRegistry` in production).
    pub secrets: Arc<dyn crate::http_request_policy::SecretResolver>,
    /// The TOML config section type for error messages
    pub config_section_hint: ConfigSectionHint,
}

#[derive(derive_more::Debug)]
pub struct ActivityWorkerCompiled {
    #[debug(skip)]
    engine: Arc<Engine>,
    #[debug(skip)]
    instance_pre: InstancePre<ActivityCtx>,
    exim: ExIm,
    #[debug(skip)]
    clock_fn: Box<dyn ClockFn>,
    #[debug(skip)]
    sleep: Arc<dyn Sleep>,
    exported_ffqn_to_index: hashbrown::HashMap<FunctionFqn, ComponentExportIndex>,
    config: ActivityConfig,
}
impl ActivityWorkerCompiled {
    pub fn new_with_config(
        runnable_component: RunnableComponent,
        config: ActivityConfig,
        engine: Arc<Engine>,
        clock_fn: Box<dyn ClockFn>,
        sleep: Arc<dyn Sleep>,
    ) -> Result<Self, WasmFileError> {
        let mut linker = wasmtime::component::Linker::new(&engine);
        // wasi
        wasmtime_wasi::p2::add_to_linker_async(&mut linker)
            .map_err(|err| WasmFileError::linking_error("cannot link wasi", err))?;
        // wasi-http
        wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)
            .map_err(|err| WasmFileError::linking_error("cannot link wasi-http", err))?;
        // obelisk:log
        log_activities::obelisk::log::log::add_to_linker::<_, ActivityCtx>(&mut linker, |x| x)
            .map_err(|err| WasmFileError::linking_error("cannot link obelisk:log", err))?;
        // Attempt to pre-instantiate to catch missing imports
        let instance_pre = linker
            .instantiate_pre(&runnable_component.wasmtime_component)
            .map_err(|err| WasmFileError::linking_error("cannot link activity", err))?;

        let exported_ffqn_to_index = RunnableComponent::index_exported_functions(
            &runnable_component.wasmtime_component,
            &runnable_component.wasm_component.exim,
        )
        .map_err(WasmFileError::DecodeError)?;
        Ok(Self {
            engine,
            exim: runnable_component.wasm_component.exim,
            clock_fn,
            sleep,
            exported_ffqn_to_index,
            config,
            instance_pre,
        })
    }

    #[must_use]
    pub fn exported_functions_ext(&self) -> &[FunctionMetadata] {
        self.exim.get_exports(true)
    }

    #[must_use]
    pub fn exports_hierarchy_ext(&self) -> &[PackageIfcFns] {
        self.exim.get_exports_hierarchy_ext()
    }

    #[must_use]
    pub fn imported_functions(&self) -> &[FunctionMetadata] {
        &self.exim.imports_flat
    }

    #[must_use]
    pub fn into_worker(
        self,
        cancel_registry: CancelRegistry,
        log_forwarder_sender: &mpsc::Sender<LogInfoAppendRow>,
        logs_storage_config: Option<LogStrageConfig>,
    ) -> ActivityWorker {
        let stdout = StdOutputConfigWithSender::new(
            self.config.forward_stdout,
            log_forwarder_sender,
            LogStreamType::StdOut,
        );
        let stderr = StdOutputConfigWithSender::new(
            self.config.forward_stderr,
            log_forwarder_sender,
            LogStreamType::StdErr,
        );
        ActivityWorker {
            engine: self.engine,
            instance_pre: self.instance_pre,
            exim: self.exim,
            clock_fn: self.clock_fn,
            sleep: self.sleep,
            exported_ffqn_to_index: self.exported_ffqn_to_index,
            config: self.config,
            cancel_registry,
            stdout,
            stderr,
            logs_storage_config,
        }
    }
}

pub struct ActivityWorker {
    engine: Arc<Engine>,
    instance_pre: InstancePre<ActivityCtx>,
    exim: ExIm,
    clock_fn: Box<dyn ClockFn>,
    sleep: Arc<dyn Sleep>,
    exported_ffqn_to_index: hashbrown::HashMap<FunctionFqn, ComponentExportIndex>,
    config: ActivityConfig,
    cancel_registry: CancelRegistry,
    stdout: Option<StdOutputConfigWithSender>,
    stderr: Option<StdOutputConfigWithSender>,
    logs_storage_config: Option<LogStrageConfig>,
}

impl ActivityWorker {
    #[must_use]
    pub fn exported_functions_ext(&self) -> &[FunctionMetadata] {
        self.exim.get_exports(true)
    }

    #[must_use]
    pub fn exports_hierarchy_ext(&self) -> &[PackageIfcFns] {
        self.exim.get_exports_hierarchy_ext()
    }

    #[must_use]
    pub fn imported_functions(&self) -> &[FunctionMetadata] {
        &self.exim.imports_flat
    }
}

#[async_trait]
impl Worker for ActivityWorker {
    fn exported_functions_noext(&self) -> &[FunctionMetadata] {
        self.exim.get_exports(false)
    }

    async fn run(&self, ctx: WorkerContext) -> WorkerResult {
        trace!("Context: {ctx:?}");
        assert!(ctx.event_history.is_empty());
        let started_at = self.clock_fn.now();
        ctx.worker_span.record(
            "execution_deadline",
            tracing::field::display(&ctx.locked_event.lock_expires_at),
        );

        let ffqn = ctx.ffqn.clone();
        let params = ctx.params.clone();
        let version = ctx.version.clone();
        let worker_span = ctx.worker_span.clone();

        let cancellation_token = self
            .cancel_registry
            .activity_obtain_cancellation_token(ctx.execution_id.clone());
        let mut execution_interrupt_watcher = ctx.execution_interrupt_watcher.clone();

        let (mut store, deadline_duration) = match self.create_store(ctx, started_at) {
            Ok(store) => store,
            Err(err) => return WorkerResult::Err(err),
        };

        let stopwatch_for_reporting = now_tokio_instant(); // Not using `clock_fn` here is ok, value is only used for log reporting.

        let call_function = {
            let call_func_params = match self
                .call_func_params(&ffqn, &params, &version, &mut store)
                .await
            {
                Ok(ok) => ok,
                Err(err) => return WorkerResult::Err(err),
            };
            self.call_func(&mut store, call_func_params)
        };

        tokio::select! { // future's liveness: Dropping the loser immediately.
            res = call_function => {
                let activity_ctx = store.into_data();
                let res = self.process_res(res, &version, activity_ctx);
                worker_span.in_scope(|| {
                    match &res {
                        Ok(worker_res_ok) => {
                            info!(duration = ?stopwatch_for_reporting.elapsed(), "Run finished: {worker_res_ok}");
                        }
                        Err(WorkerError::ExecutionYielded {
                            reason: executor::worker::ExecutionYieldReason::ExecutorClosing,
                            ..
                        }) => {
                            info!("Executor closing");
                        }
                        Err(err) => {
                            info!(%err, duration = ?stopwatch_for_reporting.elapsed(), "Run finished with an error");
                        }
                    }
                });
                return res;
            },
            ()  = self.sleep.sleep(deadline_duration) => {
                let activity_ctx = store.into_data();
                worker_span.in_scope(||
                        info!(duration = ?stopwatch_for_reporting.elapsed(), %started_at,
                        now = %self.clock_fn.now(),
                        "Run timed out")
                    );
                let http_client_traces = Some(activity_ctx.http_hooks.http_client_traces
                    .into_iter()
                        .map(|(req, mut resp)| HttpClientTrace {
                            req,
                            resp: resp.try_recv().ok(),
                        })
                        .collect_vec());
                return WorkerResult::Err(WorkerError::TemporaryTimeout{
                    http_client_traces,
                    version,
                });
            }
            _ = cancellation_token => {
                // The token fires only on cancellation (CancelRegistry::cancel_activity is its
                // sole trigger). Return Cancelled; dropping the store here tears down the
                // invocation, and the executor appends the terminal only if still `cancelling`.
                // TODO: Add http traces
                debug!("Activity run interrupted, finalizing cancellation after dropping store");
                return WorkerResult::Err(WorkerError::FatalError(FatalError::Cancelled, version));
            }
            _ = execution_interrupt_watcher.changed() => {
                debug!("Executor closing");
                // Executor will append the Unlocked event.
                return WorkerResult::Err(WorkerError::ExecutionYielded {
                    version: version.clone(),
                    reason: executor::worker::ExecutionYieldReason::ExecutorClosing,
                })
            }
        }
    }
}

struct CallFuncParams {
    func: wasmtime::component::Func,
    params: Arc<[Val]>,
    result_type: Type,
}

impl ActivityWorker {
    fn create_store(
        &self,
        ctx: WorkerContext,
        started_at: DateTime<Utc>,
    ) -> Result<(Store<ActivityCtx>, Duration /* deadline duration*/), WorkerError> {
        let lock_expires_at = ctx.locked_event.lock_expires_at;
        let worker_span = ctx.worker_span.clone();
        let version = ctx.version.clone();

        let stdout = self
            .stdout
            .as_ref()
            .map(|it| it.build(&ctx.execution_id, ctx.locked_event.run_id));
        let stderr = self
            .stderr
            .as_ref()
            .map(|it| it.build(&ctx.execution_id, ctx.locked_event.run_id));

        let mut store = activity_ctx::store(
            &self.engine,
            ctx,
            &self.config,
            self.clock_fn.clone_box(),
            stdout,
            stderr,
            self.logs_storage_config.clone(),
        );

        // Set fuel.
        if let Some(fuel) = self.config.fuel {
            store
                .set_fuel(fuel)
                .expect("engine must have `consume_fuel` enabled");
        }

        // Configure epoch callback to yield to tokio periodically.
        // Unlike `epoch_deadline_async_yield_and_update` (which yields via wasmtime's
        // runtime-agnostic yield), `tokio::task::yield_now()` cooperates with tokio's
        // task budget so the fiber yields fairly to other tokio tasks on the worker.
        store.epoch_deadline_callback(|_store_ctx| {
            Ok(wasmtime::UpdateDeadline::YieldCustom(
                1,
                Box::pin(tokio::task::yield_now()),
            ))
        });

        let deadline_delta = lock_expires_at - started_at;
        let Ok(deadline_duration) = deadline_delta.to_std() else {
            worker_span.in_scope(|| {
                info!(execution_deadline = %lock_expires_at, %started_at,
                    "Timed out - started_at later than execution_deadline");
            });
            return Err(WorkerError::TemporaryTimeout {
                http_client_traces: None,
                version,
            });
        };
        worker_span.record(
            "deadline_duration",
            tracing::field::debug(&deadline_duration),
        );
        Ok((store, deadline_duration))
    }

    async fn call_func_params(
        &self,
        ffqn: &FunctionFqn,
        params: &Params,
        version: &Version,
        store: &mut Store<ActivityCtx>,
    ) -> Result<CallFuncParams, WorkerError> {
        let instance = match self.instance_pre.instantiate_async(&mut *store).await {
            Ok(instance) => instance,
            Err(err) => {
                let reason = err.to_string();
                if reason.starts_with("maximum concurrent") {
                    return Err(WorkerError::LimitReached {
                        reason,
                        version: version.clone(),
                    });
                }
                return Err(WorkerError::FatalError(
                    FatalError::CannotInstantiate {
                        reason: format!("cannot instantiate: {err}"),
                        detail: Some(format!("{err:?}")),
                    },
                    version.clone(),
                ));
            }
        };
        let func = {
            let fn_export_index = self
                .exported_ffqn_to_index
                .get(ffqn)
                .expect("executor only calls `run` with ffqns that are exported");
            instance
                .get_func(&mut *store, fn_export_index)
                .expect("exported function found with wit-parser but not with wasmtime")
        };

        let component_func = func.ty(store);
        let params = match params.as_vals(component_func.params()) {
            Ok(params) => params,
            Err(err) => {
                return Err(WorkerError::FatalError(
                    FatalError::ParamsParsingError(err),
                    version.clone(),
                ));
            }
        };

        let result_types = component_func.results().collect::<Vec<_>>(); // TODO: investigate using the iterator directly.
        assert!(
            result_types.len() == 1,
            "multi-value and void results are not supported, must have been checked in function registry"
        );

        Ok(CallFuncParams {
            func,
            params,
            result_type: result_types
                .into_iter()
                .next()
                .expect("just checked that size == 1"),
        })
    }

    async fn call_func(
        &self,
        store: &mut Store<ActivityCtx>,
        CallFuncParams {
            func,
            params,
            result_type,
        }: CallFuncParams,
    ) -> Result<Result<SupportedFunctionReturnValue, ResultParsingError>, wasmtime::Error> {
        let mut results = vec![Val::Bool(false)];
        let res = func
            .call_async(&mut *store, &params, &mut results)
            .await
            .map(|()| {
                (
                    results.into_iter().next().expect("results size is 1"),
                    result_type,
                )
            });
        res.map(|(val, r#type)| SupportedFunctionReturnValue::new(val, r#type))
    }

    fn process_res(
        &self,
        res: Result<Result<SupportedFunctionReturnValue, ResultParsingError>, wasmtime::Error>,
        version: &Version,
        activity_ctx: ActivityCtx,
    ) -> WorkerResult {
        let http_client_traces = Some(
            activity_ctx
                .http_hooks
                .http_client_traces
                .into_iter()
                .map(|(req, mut resp)| HttpClientTrace {
                    req,
                    resp: resp.try_recv().ok(),
                })
                .collect_vec(),
        );
        match res {
            Ok(Ok(result)) => WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                retval: result,
                version: version.clone(),
                http_client_traces,
            })),
            Ok(Err(result_parsing_err)) => WorkerResult::Err(WorkerError::FatalError(
                FatalError::ResultParsingError(result_parsing_err),
                version.clone(),
            )),
            Err(err) => WorkerResult::Err(
                if let Some(trap) = err
                    .source()
                    .and_then(|source| source.downcast_ref::<wasmtime::Trap>())
                {
                    if *trap == wasmtime::Trap::OutOfFuel {
                        WorkerError::ActivityTrap {
                            reason: format!(
                                "total fuel consumed: {}",
                                self.config
                                    .fuel
                                    .expect("must have been set as it was the reason of trap")
                            ),
                            detail: None,
                            trap_kind: TrapKind::OutOfFuel,
                            version: version.clone(),
                            http_client_traces,
                        }
                    } else {
                        WorkerError::ActivityTrap {
                            reason: trap.to_string(),
                            detail: Some(format!("{err:?}")),
                            trap_kind: TrapKind::Trap,
                            version: version.clone(),
                            http_client_traces,
                        }
                    }
                } else {
                    WorkerError::ActivityTrap {
                        reason: err.to_string(),
                        trap_kind: TrapKind::HostFunctionError,
                        detail: Some(format!("{err:?}")),
                        version: version.clone(),
                        http_client_traces,
                    }
                },
            ),
        }
    }
}

#[cfg(any(test, feature = "test"))]
pub mod test {
    use concepts::{ComponentId, ComponentType, StrVariant, component_id::ComponentDigest};
    use utils::sha256sum::calculate_sha256_file;
    use wasmtime::Engine;

    use crate::{
        RunnableComponent,
        engines::{EngineConfig, Engines},
    };

    pub async fn compile_activity(wasm_path: &str) -> (RunnableComponent, ComponentId) {
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        compile_activity_with_engine(wasm_path, &engine, ComponentType::Activity).await
    }

    #[allow(dead_code)] // falsly positive
    pub(crate) async fn compile_activity_stub(wasm_path: &str) -> (RunnableComponent, ComponentId) {
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        compile_activity_with_engine(wasm_path, &engine, ComponentType::ActivityStub).await
    }

    pub(crate) async fn compile_activity_with_engine(
        wasm_path: &str,
        engine: &Engine,
        component_type: ComponentType,
    ) -> (RunnableComponent, ComponentId) {
        assert!(component_type.is_activity());
        let file_digest = calculate_sha256_file(wasm_path).await.unwrap();
        let component_id = ComponentId::new(
            component_type,
            StrVariant::empty(),
            ComponentDigest(file_digest.0),
        )
        .unwrap();
        (
            RunnableComponent::new(wasm_path, engine, component_type).unwrap(),
            component_id,
        )
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::activity::activity_worker::test::compile_activity_with_engine;
    use crate::engines::PoolingOptions;
    use crate::engines::{EngineConfig, Engines};
    use crate::http_hooks::ConfigSectionHint;
    use crate::http_request_policy::{AllowedHostConfig, HostPattern, MethodsPattern};
    use assert_matches::assert_matches;
    use concepts::prefixed_ulid::{DEPLOYMENT_ID_DUMMY, RunId};
    use concepts::storage::Locked;
    use concepts::storage::http_client_trace::{RequestTrace, ResponseTrace};
    use concepts::storage::{DbPool, TimeoutOutcome};
    use concepts::storage::{ExecutionRequest, Version};
    use concepts::time::TokioSleep;
    use concepts::{ComponentRetryConfig, ComponentType};
    use concepts::{
        ExecutionFailureKind, FinishedExecutionFailure, SUPPORTED_RETURN_VALUE_OK_EMPTY,
    };
    use concepts::{
        ExecutionId, FunctionFqn, Params, SupportedFunctionReturnValue, prefixed_ulid::ExecutorId,
        storage::CreateRequest, storage::DbPoolCloseable,
    };
    use db_tests::Database;
    use executor::executor::LockingStrategy;
    use executor::executor::{ExecConfig, ExecTask};
    use insta::assert_json_snapshot;
    use rstest::rstest;
    use serde_json::json;
    use std::future;
    use std::time::Duration;
    use test_utils::env_or_default;
    use test_utils::sim_clock::SimClock;
    use tracing::{debug, info, info_span};
    use val_json::{
        type_wrapper::TypeWrapper,
        wast_val::{WastVal, WastValWithType},
    };

    pub(crate) type ExecTaskAndClose = (ExecTask, tokio::sync::watch::Sender<bool>);

    pub const SLEEP_LOOP_ACTIVITY_FFQN: FunctionFqn = FunctionFqn::new_static_tuple(
        test_programs_sleep_activity_builder::exports::testing::sleep::sleep::SLEEP_LOOP,
    ); // sleep-loop: func(millis: u64, iterations: u32);
    pub const HTTP_GET_SUCCESSFUL_ACTIVITY: FunctionFqn = FunctionFqn::new_static_tuple(
        test_programs_http_get_activity_builder::exports::testing::http::http_get::GET_SUCCESSFUL,
    );

    pub const FIBO_ACTIVITY_FFQN: FunctionFqn = FunctionFqn::new_static_tuple(
        test_programs_fibo_activity_builder::exports::testing::fibo::fibo::FIBO,
    ); // func(n: u8) -> u64;
    pub const FIBO_10_INPUT: u8 = 10;
    pub const FIBO_10_OUTPUT: u64 = 55;

    fn activity_config(component_id: ComponentId) -> ActivityConfig {
        ActivityConfig {
            component_id,
            forward_stdout: None,
            forward_stderr: None,
            env_vars: Arc::from([]),

            fuel: None,
            allowed_hosts: Arc::from([]),
            global_http_config: crate::http_request_policy::GlobalHttpConfig::default(),
            secrets: Arc::new(crate::http_request_policy::NoSecrets),
            config_section_hint: ConfigSectionHint::ActivityWasm,
        }
    }

    pub(crate) fn activity_config_allowed_host(
        component_id: ComponentId,
        allowed_host: &str,
    ) -> ActivityConfig {
        let allowed_hosts: Arc<[AllowedHostConfig]> = Arc::from(vec![AllowedHostConfig {
            pattern: HostPattern::parse_with_methods(allowed_host, MethodsPattern::AllMethods)
                .unwrap(),
            request_url_regex: None,
            secret_names: Vec::new(),
            replace_in: hashbrown::HashSet::new(),
        }]);
        ActivityConfig {
            component_id,
            forward_stdout: None,
            forward_stderr: None,
            env_vars: Arc::from([]),

            fuel: None,
            allowed_hosts: allowed_hosts.clone(),
            global_http_config: allowed_hosts.into(),
            secrets: Arc::new(crate::http_request_policy::NoSecrets),
            config_section_hint: ConfigSectionHint::ActivityWasm,
        }
    }

    pub(crate) async fn new_activity_worker(
        wasm_path: &str,
        engine: Arc<Engine>,
        clock_fn: Box<dyn ClockFn>,
        sleep: impl Sleep + 'static,
    ) -> (Arc<dyn Worker>, ComponentId) {
        new_activity_worker_with_config(wasm_path, engine, clock_fn, sleep, activity_config).await
    }

    async fn new_activity_worker_with_config(
        wasm_path: &str,
        engine: Arc<Engine>,
        clock_fn: Box<dyn ClockFn>,
        sleep: impl Sleep + 'static,
        config_fn: impl FnOnce(ComponentId) -> ActivityConfig,
    ) -> (Arc<dyn Worker>, ComponentId) {
        let cancel_registry = CancelRegistry::new();
        let (wasm_component, component_id) =
            compile_activity_with_engine(wasm_path, &engine, ComponentType::Activity).await;
        let (db_forwarder_sender, _) = mpsc::channel(1);
        (
            Arc::new(
                ActivityWorkerCompiled::new_with_config(
                    wasm_component,
                    config_fn(component_id.clone()),
                    engine,
                    clock_fn,
                    Arc::new(sleep),
                )
                .unwrap()
                .into_worker(cancel_registry, &db_forwarder_sender, None),
            ),
            component_id,
        )
    }

    pub(crate) async fn new_activity(
        db_pool: Arc<dyn DbPool>,
        wasm_path: &'static str,
        clock_fn: Box<dyn ClockFn>,
        sleep: impl Sleep + 'static,
        retry_config: ComponentRetryConfig,
        locking_strategy: LockingStrategy,
    ) -> ExecTaskAndClose {
        new_activity_with_config(
            db_pool,
            wasm_path,
            clock_fn,
            sleep,
            activity_config,
            retry_config,
            locking_strategy,
        )
        .await
    }

    pub(crate) async fn new_activity_with_config(
        db_pool: Arc<dyn DbPool>,
        wasm_path: &'static str,
        clock_fn: Box<dyn ClockFn>,
        sleep: impl Sleep + 'static,
        config_fn: impl FnOnce(ComponentId) -> ActivityConfig,
        retry_config: ComponentRetryConfig,
        locking_strategy: LockingStrategy,
    ) -> ExecTaskAndClose {
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        let (worker, component_id) = new_activity_worker_with_config(
            wasm_path,
            engine,
            clock_fn.clone_box(),
            sleep,
            config_fn,
        )
        .await;
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id,
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config,
            locking_strategy,
        };
        ExecTask::new_all_ffqns_test(worker, exec_config, clock_fn, db_pool)
    }

    pub(crate) async fn new_activity_fibo(
        db_pool: Arc<dyn DbPool>,
        clock_fn: Box<dyn ClockFn>,
        sleep: impl Sleep + 'static,
        locking_strategy: LockingStrategy,
    ) -> ExecTaskAndClose {
        new_activity(
            db_pool,
            test_programs_fibo_activity_builder::TEST_PROGRAMS_FIBO_ACTIVITY,
            clock_fn,
            sleep,
            ComponentRetryConfig::ZERO,
            locking_strategy,
        )
        .await
    }

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub(crate) enum TestRetryBehavior {
        SucceedOnRetry,
        Fail { expected_retry_err: &'static str },
    }

    /// Common test helper for testing HTTP GET retry behavior on fallible errors.
    ///
    /// This function tests that:
    /// 1. An activity that returns an error (from a 500 response) triggers a temporary failure with backoff
    /// 2. After the backoff expires, the activity is retried
    /// 3. On retry, either succeeds (if `succeed_eventually` is true) or fails permanently
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn run_http_get_retry_test(
        listener: std::net::TcpListener,
        worker: Arc<dyn Worker>,
        sim_clock: SimClock,
        ffqn: FunctionFqn,
        make_params: impl FnOnce(&str) -> Params,
        locking_strategy: LockingStrategy,
        expected_err_contains: &str,
        test_retry_behavior: TestRetryBehavior,
    ) {
        use std::ops::Deref;
        use wiremock::{
            Mock, MockServer, ResponseTemplate,
            matchers::{method, path},
        };

        const BODY: &str = "ok";
        const RETRY_EXP_BACKOFF: Duration = Duration::from_millis(10);

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

        let server_address = listener
            .local_addr()
            .expect("Failed to get server address.");
        let uri = format!("http://127.0.0.1:{port}", port = server_address.port());

        let retry_config = ComponentRetryConfig {
            max_retries: Some(1),
            retry_exp_backoff: RETRY_EXP_BACKOFF,
        };
        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config,
            locking_strategy,
        };
        let ffqns = Arc::from([ffqn.clone()]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        let params = make_params(&uri);
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: ffqn.clone(),
                params,
                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();

        let server = MockServer::builder().listener(listener).start().await;
        Mock::given(method("GET"))
            .and(path("/"))
            .respond_with(ResponseTemplate::new(500).set_body_string(BODY))
            .expect(1)
            .mount(&server)
            .await;
        debug!("started mock server on {}", server.address());

        {
            // Expect error result to be interpreted as a temporary failure
            assert_eq!(
                1,
                exec_task
                    .tick_test(sim_clock.now(), RunId::generate())
                    .await
                    .wait_for_tasks()
                    .await
                    .len()
            );
            let exec_log = db_connection.get(&execution_id).await.unwrap();

            let (reason, detail, found_expires_at, http_client_traces) = assert_matches!(
                &exec_log.last_event().event,
                ExecutionRequest::TemporarilyFailed {
                    backoff_expires_at,
                    reason,
                    detail: Some(detail),
                    http_client_traces: Some(http_client_traces)
                }
                => (reason, detail, *backoff_expires_at, http_client_traces)
            );
            assert_eq!(sim_clock.now() + RETRY_EXP_BACKOFF, found_expires_at);
            assert_eq!("activity finished with error", reason.deref());
            assert!(
                detail.contains(expected_err_contains),
                "Unexpected detail: {detail}, expected to contain: {expected_err_contains}"
            );

            assert_eq!(1, http_client_traces.len());
            let http_client_trace = http_client_traces.iter().next().unwrap();
            let (method_actual, uri_actual) = assert_matches!(
                http_client_trace,
                HttpClientTrace {
                    req: RequestTrace {
                        method,
                        sent_at: _,
                        uri
                    },
                    resp: Some(ResponseTrace {
                        status: Ok(500),
                        finished_at: _
                    })
                }
                => (method, uri)
            );
            assert_eq!("GET", method_actual);
            assert_eq!(format!("{uri}/"), *uri_actual);
            server.verify().await;
        }

        // Noop until the timeout expires
        assert_eq!(
            0,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        sim_clock.move_time_forward(RETRY_EXP_BACKOFF);

        server.reset().await;

        if test_retry_behavior == TestRetryBehavior::SucceedOnRetry {
            // Reconfigure the server, return 200
            Mock::given(method("GET"))
                .and(path("/"))
                .respond_with(ResponseTemplate::new(200).set_body_string(BODY))
                .expect(1)
                .mount(&server)
                .await;
            debug!("Reconfigured the server");
        } // otherwise return 404

        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let res = assert_matches!(exec_log.last_event().event.clone(), ExecutionRequest::Finished { retval, .. } => retval);
        let wast_val_with_type = match test_retry_behavior {
            TestRetryBehavior::SucceedOnRetry => {
                let wast_val_with_type = assert_matches!(res, SupportedFunctionReturnValue::Ok(Some(wast_val_with_type)) => wast_val_with_type);
                let val = assert_matches!(&wast_val_with_type.value, WastVal::String(val) => val);
                assert_eq!(BODY, val.deref());
                wast_val_with_type
            }
            TestRetryBehavior::Fail { expected_retry_err } => {
                let wast_val_with_type = assert_matches!(res, SupportedFunctionReturnValue::Err(Some(wast_val_with_type)) => wast_val_with_type);
                let val = assert_matches!(&wast_val_with_type.value, WastVal::String(val) => val);
                assert_eq!(expected_retry_err, val.deref());
                wast_val_with_type
            }
        };
        // check types
        assert_matches!(wast_val_with_type.r#type, TypeWrapper::String); // in both cases
        drop(db_connection);
        drop(exec_task);
        db_close.close().await;
    }

    /// Creates an activity worker with allowed host configuration.
    /// Returns the worker and the URI for the allowed host.
    pub(crate) async fn create_activity_worker_with_allowed_host(
        wasm_path: &str,
        listener: &std::net::TcpListener,
        clock_fn: Box<dyn ClockFn>,
    ) -> Arc<dyn Worker> {
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        let server_address = listener
            .local_addr()
            .expect("Failed to get server address.");
        let uri = format!("http://127.0.0.1:{port}", port = server_address.port());

        let (worker, _) =
            new_activity_worker_with_config(wasm_path, engine, clock_fn, TokioSleep, {
                let uri = uri.clone();
                move |component_id| activity_config_allowed_host(component_id, &uri)
            })
            .await;
        worker
    }

    #[rstest]
    #[tokio::test]
    async fn fibo_once(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        let (exec, _close_tx) = new_activity_fibo(
            db_pool.clone(),
            sim_clock.clone_box(),
            TokioSleep,
            locking_strategy,
        )
        .await;
        // Create an execution.
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let params = Params::from_json_values_test(vec![json!(FIBO_10_INPUT)]);
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: FIBO_ACTIVITY_FFQN,
                params,
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: exec.config.component_id.clone(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        // tick
        let executed = exec
            .tick_test_await(sim_clock.now(), RunId::generate())
            .await;
        assert_eq!(vec![execution_id.clone()], executed);
        // Check the result.
        let res = db_connection
            .wait_for_finished_result(
                &execution_id,
                Some(Box::pin(future::ready(TimeoutOutcome::Cancel))),
            )
            .await
            .unwrap();
        let res = assert_matches!(res, SupportedFunctionReturnValue::Ok(ok) => ok);
        let fibo = assert_matches!(res,
            Some(WastValWithType {value: WastVal::U64(val), r#type: TypeWrapper::U64 }) => val);
        assert_eq!(FIBO_10_OUTPUT, fibo);
        drop(db_connection);
        db_close.close().await;
    }

    #[tokio::test]
    async fn limit_reached() {
        const FIBO_INPUT: u8 = 10;
        const LOCK_EXPIRY_MILLIS: u64 = 1100;
        const TASKS: u32 = 10;
        const MAX_INSTANCES: u32 = 1;

        test_utils::set_up();
        let fibo_input = env_or_default("FIBO_INPUT", FIBO_INPUT);
        let lock_expiry =
            Duration::from_millis(env_or_default("LOCK_EXPIRY_MILLIS", LOCK_EXPIRY_MILLIS));
        let tasks = env_or_default("TASKS", TASKS);
        let max_instances = env_or_default("MAX_INSTANCES", MAX_INSTANCES);

        let pool_opts = PoolingOptions {
            pooling_total_component_instances: Some(max_instances),
            pooling_total_stacks: Some(max_instances),
            pooling_total_core_instances: Some(max_instances),
            pooling_total_memories: Some(max_instances),
            pooling_total_tables: Some(max_instances),
            ..Default::default()
        };

        let engine =
            Engines::get_activity_engine_test(EngineConfig::pooling_nocache_testing(pool_opts))
                .unwrap();
        let sim_clock = SimClock::epoch();

        let (fibo_worker, _) = new_activity_worker(
            test_programs_fibo_activity_builder::TEST_PROGRAMS_FIBO_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
        )
        .await;
        // create executions
        let join_handles = (0..tasks)
            .map(|_| {
                let fibo_worker = fibo_worker.clone();
                let sim_clock = sim_clock.clone();
                let execution_id = ExecutionId::generate();
                let (execution_interrupt_tx, execution_interrupt_watcher) =
                    tokio::sync::watch::channel(false);
                let ctx = WorkerContext {
                    execution_id: execution_id.clone(),
                    metadata: concepts::ExecutionMetadata::empty(),
                    component_digest: ComponentId::dummy_activity().component_digest,
                    ffqn: FIBO_ACTIVITY_FFQN,
                    params: Params::from_json_values_test(vec![json!(fibo_input)]),
                    event_history: Vec::new(),
                    responses: Vec::new(),
                    parent: None,
                    version: Version::new(0),
                    can_be_retried: false,
                    worker_span: info_span!("worker-test"),
                    locked_event: Locked {
                        component_id: ComponentId::dummy_activity(),
                        executor_id: ExecutorId::generate(),
                        deployment_id: DEPLOYMENT_ID_DUMMY,
                        run_id: RunId::generate(),
                        lock_expires_at: sim_clock.now() + lock_expiry,
                        retry_config: ComponentRetryConfig::ZERO,
                    },
                    execution_interrupt_watcher,
                };
                tokio::spawn(async move {
                    let res = fibo_worker.run(ctx).await;
                    drop(execution_interrupt_tx);
                    res
                })
            })
            .collect::<Vec<_>>();
        let mut limit_reached = 0;
        for jh in join_handles {
            if matches!(
                jh.await.unwrap(),
                WorkerResult::Err(WorkerError::LimitReached { .. })
            ) {
                limit_reached += 1;
            }
        }
        assert!(limit_reached > 0, "Limit was not reached");
    }

    #[rstest::rstest]
    #[case(
            10,
            100,
            SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure{
                kind: ExecutionFailureKind::TimedOut,
                reason: None, detail: None
            })
        )] // 1s -> timeout
    #[case(10, 10, SUPPORTED_RETURN_VALUE_OK_EMPTY)] // 0.1s -> Ok
    #[case(
            1500,
            1,
            SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure{
                kind: ExecutionFailureKind::TimedOut,
                reason: None, detail: None
            })
        )] // 1s -> timeout
    #[tokio::test]
    async fn sleep_should_produce_temporary_timeout(
        #[case] sleep_millis: u32,
        #[case] sleep_iterations: u32,
        #[case] expected: concepts::SupportedFunctionReturnValue,
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        const LOCK_EXPIRY: Duration = Duration::from_millis(500);
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        let (worker, _) = new_activity_worker(
            test_programs_sleep_activity_builder::TEST_PROGRAMS_SLEEP_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
        )
        .await;

        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: LOCK_EXPIRY,
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };
        let ffqns = Arc::from([SLEEP_LOOP_ACTIVITY_FFQN]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        // Create an execution.
        let execution_id = ExecutionId::generate();
        info!("Testing {execution_id}");
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection().await.unwrap();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: SLEEP_LOOP_ACTIVITY_FFQN,
                params: Params::from_json_values_test(vec![
                    json!(
                        {"milliseconds": sleep_millis}),
                    json!(sleep_iterations),
                ]),
                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();

        // Run the execution via tick.
        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );

        // Check the result.
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let retval = assert_matches!(
            exec_log.last_event().event.clone(),
            ExecutionRequest::Finished { retval, .. } => retval
        );
        assert_eq!(expected, retval);

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

    #[rstest::rstest]
    #[case(1, 2_000)] // 1ms * 2000 iterations
    #[case(2_000, 1)] // 2s * 1 iteration
    #[tokio::test]
    async fn long_running_execution_should_timeout(
        #[case] sleep_millis: u64,
        #[case] sleep_iterations: u32,
    ) {
        const TIMEOUT: Duration = Duration::from_millis(200);
        test_utils::set_up();

        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();

        let sim_clock = SimClock::epoch();
        let (worker, _) = new_activity_worker(
            test_programs_sleep_activity_builder::TEST_PROGRAMS_SLEEP_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
        )
        .await;

        let executed_at = sim_clock.now();
        let version = Version::new(10);
        let (_execution_interrupt_tx, execution_interrupt_watcher) =
            tokio::sync::watch::channel(false);
        let ctx = WorkerContext {
            execution_id: ExecutionId::generate(),
            metadata: concepts::ExecutionMetadata::empty(),
            component_digest: ComponentId::dummy_activity().component_digest,
            ffqn: SLEEP_LOOP_ACTIVITY_FFQN,
            params: Params::from_json_values_test(vec![
                json!(
                    {"milliseconds": sleep_millis}),
                json!(sleep_iterations),
            ]),
            event_history: Vec::new(),
            responses: Vec::new(),
            parent: None,
            version: version.clone(),
            can_be_retried: false,
            worker_span: info_span!("worker-test"),
            locked_event: Locked {
                component_id: ComponentId::dummy_activity(),
                executor_id: ExecutorId::generate(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                run_id: RunId::generate(),
                lock_expires_at: executed_at + TIMEOUT,
                retry_config: ComponentRetryConfig::ZERO,
            },
            execution_interrupt_watcher,
        };
        let WorkerResult::Err(err) = worker.run(ctx).await else {
            panic!()
        };
        let actual_version = assert_matches!(
            err,
            WorkerError::TemporaryTimeout {
                http_client_traces:_,
                version
            }
            => version
        );
        assert_eq!(version, actual_version);
    }

    #[tokio::test]
    async fn execution_deadline_before_now_should_timeout() {
        test_utils::set_up();

        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();
        let sim_clock = SimClock::epoch();
        let (worker, _) = new_activity_worker(
            test_programs_sleep_activity_builder::TEST_PROGRAMS_SLEEP_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
        )
        .await;
        // simulate a scheduling problem where deadline < now
        let execution_deadline = sim_clock.now();
        sim_clock.move_time_forward(Duration::from_millis(100));
        let version = Version::new(10);
        let (_execution_interrupt_tx, execution_interrupt_watcher) =
            tokio::sync::watch::channel(false);
        let ctx = WorkerContext {
            execution_id: ExecutionId::generate(),
            metadata: concepts::ExecutionMetadata::empty(),
            component_digest: ComponentId::dummy_activity().component_digest,
            ffqn: SLEEP_LOOP_ACTIVITY_FFQN,
            params: Params::from_json_values_test(vec![
                json!(
                    {"milliseconds": 1}),
                json!(1),
            ]),
            event_history: Vec::new(),
            responses: Vec::new(),
            parent: None,
            version: version.clone(),
            can_be_retried: false,
            worker_span: info_span!("worker-test"),
            locked_event: 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,
            },
            execution_interrupt_watcher,
        };
        let WorkerResult::Err(err) = worker.run(ctx).await else {
            panic!()
        };
        let actual_version = assert_matches!(
            err,
            WorkerError::TemporaryTimeout {
                http_client_traces: None,
                version: actual_version,
            }
            => actual_version
        );
        assert_eq!(version, actual_version);
    }

    #[rstest]
    #[tokio::test]
    async fn http_get_simple(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        use std::ops::Deref;
        use wiremock::{
            Mock, MockServer, ResponseTemplate,
            matchers::{header, method, path},
        };
        const BODY: &str = "ok";
        test_utils::set_up();
        info!("All set up");
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let server_address = listener
            .local_addr()
            .expect("Failed to get server address.");
        let uri = format!("http://127.0.0.1:{port}", port = server_address.port());

        let (worker, _) = new_activity_worker_with_config(
            test_programs_http_get_activity_builder::TEST_PROGRAMS_HTTP_GET_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
            {
                let uri = uri.clone();
                move |component_id| activity_config_allowed_host(component_id, &uri)
            },
        )
        .await;

        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };
        let ffqns = Arc::from([HTTP_GET_SUCCESSFUL_ACTIVITY]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        let params = Params::from_json_values_test(vec![json!(uri.clone())]);
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        info!("Creating execution");
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: HTTP_GET_SUCCESSFUL_ACTIVITY,
                params,
                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();

        let server = MockServer::builder().listener(listener).start().await;
        Mock::given(method("GET"))
            .and(path("/"))
            .and(header("host", server_address.to_string()))
            .respond_with(ResponseTemplate::new(200).set_body_string(BODY))
            .expect(1)
            .mount(&server)
            .await;

        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let (res, http_client_traces) = assert_matches!(
                exec_log.last_event().event.clone(),
                ExecutionRequest::Finished { retval, http_client_traces: Some(http_client_traces) }
                => (retval, http_client_traces));
        let wast_val_with_type = assert_matches!(res, SupportedFunctionReturnValue::Ok(Some(wast_val_with_type)) => wast_val_with_type);
        let val = assert_matches!(wast_val_with_type.value, WastVal::String(val) => val);
        assert_eq!(BODY, val.deref());
        // check types
        assert_matches!(wast_val_with_type.r#type, TypeWrapper::String);
        assert_eq!(1, http_client_traces.len());
        let http_client_trace = http_client_traces.into_iter().next().unwrap();
        let (method, uri_actual) = assert_matches!(
            http_client_trace,
            HttpClientTrace {
                req: RequestTrace {
                    method,
                    sent_at: _,
                    uri
                },
                resp: Some(ResponseTrace {
                    status: Ok(200),
                    finished_at: _
                })
            }
            => (method, uri)
        );
        assert_eq!("GET", method);
        assert_eq!(format!("{uri}/"), *uri_actual);
        drop(db_connection);
        drop(exec_task);
        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn http_get_activity_trap_should_be_turned_into_finished_execution_error_permanent_failure(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        use wiremock::{
            Mock, MockServer, ResponseTemplate,
            matchers::{method, path},
        };
        const STATUS: u16 = 418; // I'm a teapot causes trap
        test_utils::set_up();
        info!("All set up");
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let server_address = listener
            .local_addr()
            .expect("Failed to get server address.");
        let uri = format!("http://127.0.0.1:{port}", port = server_address.port());

        let (worker, _) = new_activity_worker_with_config(
            test_programs_http_get_activity_builder::TEST_PROGRAMS_HTTP_GET_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
            {
                let uri = uri.clone();
                move |component_id| activity_config_allowed_host(component_id, &uri)
            },
        )
        .await;

        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };
        let ffqns = Arc::from([HTTP_GET_SUCCESSFUL_ACTIVITY]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        let params = Params::from_json_values_test(vec![json!(uri.clone())]);
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        info!("Creating execution");
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: HTTP_GET_SUCCESSFUL_ACTIVITY,
                params,
                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();

        let server = MockServer::builder().listener(listener).start().await;
        Mock::given(method("GET"))
            .and(path("/"))
            .respond_with(ResponseTemplate::new(STATUS).set_body_string(""))
            .expect(1)
            .mount(&server)
            .await;

        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let (res, http_client_traces) = assert_matches!(
                exec_log.last_event().event.clone(),
                ExecutionRequest::Finished { retval, http_client_traces: Some(http_client_traces) }
                => (retval, http_client_traces));
        let res = assert_matches!(res, SupportedFunctionReturnValue::ExecutionFailure(err) => err);
        let reason = assert_matches!(
            res,
            FinishedExecutionFailure {
                kind: ExecutionFailureKind::Uncategorized,
                reason: Some(reason), // activity trap
                detail: _
            } => reason
        );
        assert!(reason.starts_with("activity trap"), "{reason}");

        assert_eq!(1, http_client_traces.len());
        let http_client_trace = http_client_traces.into_iter().next().unwrap();
        let (method, uri_actual) = assert_matches!(
            http_client_trace,
            HttpClientTrace {
                req: RequestTrace {
                    method,
                    sent_at: _,
                    uri
                },
                resp: Some(ResponseTrace {
                    status: Ok(STATUS),
                    finished_at: _
                })
            }
            => (method, uri)
        );
        assert_eq!("GET", method);
        assert_eq!(format!("{uri}/"), *uri_actual);
        drop(db_connection);
        drop(exec_task);
        db_close.close().await;
    }

    #[rstest::rstest]
    #[tokio::test]
    async fn http_get_retry_on_fallible_err(
        #[values(TestRetryBehavior::SucceedOnRetry,TestRetryBehavior::Fail { expected_retry_err: "wrong status code: 404" })]
        test_retry_behavior: TestRetryBehavior,
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        test_utils::set_up();

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let sim_clock = SimClock::epoch();
        let worker = create_activity_worker_with_allowed_host(
            test_programs_http_get_activity_builder::TEST_PROGRAMS_HTTP_GET_ACTIVITY,
            &listener,
            sim_clock.clone_box(),
        )
        .await;

        run_http_get_retry_test(
            listener,
            worker,
            sim_clock,
            HTTP_GET_SUCCESSFUL_ACTIVITY,
            |uri| Params::from_json_values_test(vec![json!(uri)]),
            locking_strategy,
            "wrong status code: 500",
            test_retry_behavior,
        )
        .await;
    }

    #[tokio::test]
    async fn http_get_denied_host() {
        use wiremock::{
            Mock, MockServer, ResponseTemplate,
            matchers::{method, path},
        };
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let server_address = listener.local_addr().unwrap();
        let uri = format!("http://127.0.0.1:{port}", port = server_address.port());

        // Create worker with NO allowed hosts - the request should be denied
        let (worker, _) = new_activity_worker_with_config(
            test_programs_http_get_activity_builder::TEST_PROGRAMS_HTTP_GET_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
            activity_config, // no allowed hosts
        )
        .await;

        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: ComponentId::dummy_activity(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy: LockingStrategy::ByComponentDigest,
        };
        let ffqns = Arc::from([HTTP_GET_SUCCESSFUL_ACTIVITY]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        let params = Params::from_json_values_test(vec![json!(uri.clone())]);
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: HTTP_GET_SUCCESSFUL_ACTIVITY,
                params,
                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();

        let server = MockServer::builder().listener(listener).start().await;
        Mock::given(method("GET"))
            .and(path("/"))
            .respond_with(ResponseTemplate::new(200).set_body_string("should not reach"))
            .expect(0) // Should NOT be called since host is denied
            .mount(&server)
            .await;

        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let retval = assert_matches!(
            exec_log.last_event().event.clone(),
            ExecutionRequest::Finished { retval, .. } => retval
        );
        // The execution should fail with an ExecutionFailure when the WASM traps.
        // The trap happens because the HTTP request is denied and the WASM unwraps the error.
        let err = assert_matches!(retval, SupportedFunctionReturnValue::Err(Some(err)) => err);
        let err = assert_matches!(err.value, WastVal::String(err) => err);
        assert_eq!("ErrorCode::HttpRequestDenied", err);
        // Verify the mock server was not called (request was blocked before reaching it)
        server.verify().await;
        drop(db_connection);
        drop(exec_task);
        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn http_get_with_secret(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        use crate::http_request_policy::{AllowedHostConfig, MethodsPattern, ReplacementLocation};
        use hashbrown::HashSet;
        use secrecy::SecretString;
        use wiremock::{
            Mock, MockServer, ResponseTemplate,
            matchers::{header, method, path, query_param},
        };
        const SECRET_VALUE: &str = "my-secret-api-key-12345";
        const SECRET_ENV_VAR: &str = "TEST_API_KEY";
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let engine = Engines::get_activity_engine_test(EngineConfig::on_demand_testing()).unwrap();

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let server_address = listener.local_addr().unwrap();
        let allowed_host = format!("http://127.0.0.1:{port}", port = server_address.port());
        let host_pattern =
            HostPattern::parse_with_methods(&allowed_host, MethodsPattern::AllMethods).unwrap();

        // Create worker with secret configuration
        let (worker, component_id) = new_activity_worker_with_config(
            test_programs_http_get_activity_builder::TEST_PROGRAMS_HTTP_GET_ACTIVITY,
            engine,
            sim_clock.clone_box(),
            TokioSleep,
            {
                let host_pattern = host_pattern.clone();
                move |component_id| {
                    let allowed_hosts: Arc<[AllowedHostConfig]> =
                        Arc::from(vec![AllowedHostConfig {
                            pattern: host_pattern,
                            request_url_regex: None,
                            secret_names: vec![SECRET_ENV_VAR.to_string()],
                            replace_in: HashSet::from_iter([
                                ReplacementLocation::Headers,
                                ReplacementLocation::Params,
                                ReplacementLocation::Body,
                            ]),
                        }]);
                    let secrets = Arc::new(crate::http_request_policy::TestSecretResolver(
                        hashbrown::HashMap::from_iter([(
                            SECRET_ENV_VAR.to_string(),
                            SecretString::from(SECRET_VALUE.to_string()),
                        )]),
                    ));
                    ActivityConfig {
                        component_id,
                        forward_stdout: None,
                        forward_stderr: None,
                        env_vars: Arc::from([]),
                        fuel: None,
                        allowed_hosts: allowed_hosts.clone(),
                        global_http_config: allowed_hosts.into(),
                        secrets,
                        config_section_hint: ConfigSectionHint::ActivityWasm,
                    }
                }
            },
        )
        .await;

        let exec_config = ExecConfig {
            batch_size: 1,
            lock_expiry: Duration::from_secs(1),
            tick_sleep: Duration::ZERO,
            component_id: component_id.clone(),
            task_limiter_global: None,
            task_limiter_local: None,
            executor_id: ExecutorId::generate(),
            retry_config: ComponentRetryConfig::ZERO,
            locking_strategy,
        };
        let secret_get_ffqn: FunctionFqn =
            FunctionFqn::new_static("testing:http/http-get", "secret-get");
        let ffqns = Arc::from([secret_get_ffqn.clone()]);
        let (exec_task, _close_tx) = ExecTask::new_test(
            exec_config,
            worker,
            sim_clock.clone_box(),
            db_pool.clone(),
            ffqns,
        );

        // secret-get takes: url, env_var, header (optional)
        // The url contains the placeholder which gets replaced with the secret
        let url_with_placeholder = format!("{allowed_host}/?secret={SECRET_ENV_VAR}");
        let header_with_placeholder =
            Some(("X-API-Key".to_string(), format!("Bearer {SECRET_ENV_VAR}")));
        let params = Params::from_json_values_test(vec![
            json!(url_with_placeholder),
            json!(SECRET_ENV_VAR),
            json!(header_with_placeholder),
        ]);
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: secret_get_ffqn.clone(),
                params,
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id,
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();

        let server = MockServer::builder().listener(listener).start().await;
        // Verify the secret was replaced in both query params and headers
        Mock::given(method("GET"))
            .and(path("/"))
            .and(query_param("secret", SECRET_VALUE))
            .and(header("X-API-Key", format!("Bearer {SECRET_VALUE}")))
            .respond_with(ResponseTemplate::new(200).set_body_string("secret-received"))
            .expect(1)
            .mount(&server)
            .await;

        assert_eq!(
            1,
            exec_task
                .tick_test(sim_clock.now(), RunId::generate())
                .await
                .wait_for_tasks()
                .await
                .len()
        );
        let exec_log = db_connection.get(&execution_id).await.unwrap();
        let retval = assert_matches!(
            exec_log.last_event().event.clone(),
            ExecutionRequest::Finished { retval, .. } => retval
        );
        // Should succeed with the secret replaced
        assert_matches!(retval, SupportedFunctionReturnValue::Ok(..));
        server.verify().await;
        drop(db_connection);
        drop(exec_task);
        db_close.close().await;
    }

    #[rstest::rstest(
            param => [
                r#"{"image": "foo", "a": false, "b":false}"#,
                r#"{"b": false, "a":false, "image": "foo"}"#,
                ])]
    #[tokio::test]
    async fn record_field_ordering(
        param: &str,
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        let (exec, _close_tx) = new_activity_with_config(
            db_pool.clone(),
            test_programs_serde_activity_builder::TEST_PROGRAMS_SERDE_ACTIVITY,
            sim_clock.clone_box(),
            TokioSleep,
            move |component_id| ActivityConfig {
                component_id,
                forward_stdout: Some(StdOutputConfig::Stderr),
                forward_stderr: Some(StdOutputConfig::Stderr),
                env_vars: Arc::default(),

                fuel: None,
                allowed_hosts: Arc::from([]),
                global_http_config: crate::http_request_policy::GlobalHttpConfig::default(),
                secrets: Arc::new(crate::http_request_policy::NoSecrets),
                config_section_hint: ConfigSectionHint::ActivityWasm,
            },
            ComponentRetryConfig::ZERO,
            locking_strategy,
        )
        .await;
        // Create an execution.
        let ffqn = FunctionFqn::new_static_tuple(
            test_programs_serde_activity_builder::exports::testing::serde::serde::REC,
        );
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn,
                params: Params::from_json_values_test(vec![serde_json::from_str(param).unwrap()]),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: exec.config.component_id.clone(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let executed = exec
            .tick_test_await(sim_clock.now(), RunId::generate())
            .await;
        assert_eq!(vec![execution_id.clone()], executed);
        // Check the result.
        let res = db_connection
            .wait_for_finished_result(
                &execution_id,
                Some(Box::pin(future::ready(TimeoutOutcome::Cancel))),
            )
            .await
            .unwrap();
        let record = assert_matches!(res, SupportedFunctionReturnValue::Ok(record) => record);
        insta::with_settings!({
            prepend_module_to_snapshot => false},
            {
                assert_json_snapshot!(record);
            }
        );
        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn variant_with_optional_none(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        let (exec, _close_tx) = new_activity_with_config(
            db_pool.clone(),
            test_programs_serde_activity_builder::TEST_PROGRAMS_SERDE_ACTIVITY,
            sim_clock.clone_box(),
            TokioSleep,
            move |component_id| ActivityConfig {
                component_id,
                forward_stdout: Some(StdOutputConfig::Stderr),
                forward_stderr: Some(StdOutputConfig::Stderr),
                env_vars: Arc::default(),

                fuel: None,
                allowed_hosts: Arc::from([]),
                global_http_config: crate::http_request_policy::GlobalHttpConfig::default(),
                secrets: Arc::new(crate::http_request_policy::NoSecrets),
                config_section_hint: ConfigSectionHint::ActivityWasm,
            },
            ComponentRetryConfig::ZERO,
            locking_strategy,
        )
        .await;
        // Create an execution.
        let ffqn = FunctionFqn::new_static_tuple(
            test_programs_serde_activity_builder::exports::testing::serde::serde::VAR,
        );
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn,
                params: Params::from_json_values_test(vec![json!({"var1":null})]),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: exec.config.component_id.clone(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let executed = exec
            .tick_test_await(sim_clock.now(), RunId::generate())
            .await;
        assert_eq!(vec![execution_id.clone()], executed);
        // Check the result.
        let res = db_connection
            .wait_for_finished_result(
                &execution_id,
                Some(Box::pin(future::ready(TimeoutOutcome::Cancel))),
            )
            .await
            .unwrap();
        let variant = assert_matches!(res, SupportedFunctionReturnValue::Ok(variant) => variant);

        insta::with_settings!({
            prepend_module_to_snapshot => false},
            {
                assert_json_snapshot!(variant);
            }
        );
        db_close.close().await;
    }

    #[rstest]
    #[tokio::test]
    async fn permanent_error_variant_should_not_retry(
        #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
        locking_strategy: LockingStrategy,
    ) {
        test_utils::set_up();
        let sim_clock = SimClock::default();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let db_connection = db_pool.connection().await.unwrap();
        // max_retries > 0, so it would retry if error wasn't permanent
        let retry_config = ComponentRetryConfig {
            max_retries: Some(1),
            retry_exp_backoff: Duration::from_millis(10),
        };
        let (exec, _close_tx) = new_activity_with_config(
            db_pool.clone(),
            test_programs_serde_activity_builder::TEST_PROGRAMS_SERDE_ACTIVITY,
            sim_clock.clone_box(),
            TokioSleep,
            move |component_id| ActivityConfig {
                component_id,
                forward_stdout: Some(StdOutputConfig::Stderr),
                forward_stderr: Some(StdOutputConfig::Stderr),
                env_vars: Arc::default(),

                fuel: None,
                allowed_hosts: Arc::from([]),
                global_http_config: crate::http_request_policy::GlobalHttpConfig::default(),
                secrets: Arc::new(crate::http_request_policy::NoSecrets),
                config_section_hint: ConfigSectionHint::ActivityWasm,
            },
            retry_config,
            locking_strategy,
        )
        .await;
        // Create an execution.
        let ffqn = FunctionFqn::new_static_tuple(
            test_programs_serde_activity_builder::exports::testing::serde::serde::PERMANENT_ERR,
        );
        let execution_id = ExecutionId::generate();
        let created_at = sim_clock.now();
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: exec.config.component_id.clone(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let executed = exec
            .tick_test_await(sim_clock.now(), RunId::generate())
            .await;
        assert_eq!(vec![execution_id.clone()], executed);
        // Check the result - should be Finished with Err, not TemporarilyFailed
        let res = db_connection
            .wait_for_finished_result(
                &execution_id,
                Some(Box::pin(future::ready(TimeoutOutcome::Cancel))),
            )
            .await
            .unwrap();
        // The permanent-failure variant should prevent retry and finish with Err
        let err = assert_matches!(res, SupportedFunctionReturnValue::Err(err) => err);
        let (key, _) = assert_matches!(
            err,
            Some(WastValWithType {
                value: WastVal::Variant(key, payload),
                ..
            }) => (key, payload)
        );
        assert_eq!("permanent_failure", key.as_snake_str());
        db_close.close().await;
    }
}