obeli_sk_concepts/

storage.rs

use crate::ComponentId;
use crate::ComponentRetryConfig;
use crate::ComponentType;
use crate::ContentDigest;
use crate::ExecutionFailureKind;
use crate::ExecutionId;
use crate::ExecutionMetadata;
use crate::FunctionExtension;
use crate::FunctionFqn;
use crate::FunctionMetadata;
use crate::JoinSetId;
use crate::Params;
use crate::StrVariant;
use crate::SupportedFunctionReturnValue;
use crate::component_id::ComponentDigest;
use crate::prefixed_ulid::DelayId;
use crate::prefixed_ulid::DeploymentId;
use crate::prefixed_ulid::ExecutionIdDerived;
use crate::prefixed_ulid::ExecutorId;
use crate::prefixed_ulid::RunId;
use assert_matches::assert_matches;
use async_trait::async_trait;
use chrono::TimeDelta;
use chrono::{DateTime, Utc};
use http_client_trace::HttpClientTrace;
use serde::Deserialize;
use serde::Serialize;
use std::fmt::Debug;
use std::fmt::Display;
use std::num::NonZeroU16;
use std::panic::Location;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tracing::instrument;
use tracing_error::SpanTrace;

// Shared between databases. TODO: Extract to db-common
pub const STATE_PENDING_AT: &str = "pending_at";
pub const STATE_BLOCKED_BY_JOIN_SET: &str = "blocked_by_join_set";
pub const STATE_LOCKED: &str = "locked";
pub const STATE_FINISHED: &str = "finished";
// `lifecycle` column values on `t_state`: an override of the underlying pending
// state. Mutually exclusive by construction (single column).
pub const LIFECYCLE_ACTIVE: &str = "active";
pub const LIFECYCLE_PAUSED: &str = "paused";
pub const LIFECYCLE_CANCELLING: &str = "cancelling";

/// Typed view of the `t_state.lifecycle` column. Mutually exclusive by
/// construction (single column), so pause and cancellation cannot coexist.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Lifecycle {
    Active,
    Paused,
    Cancelling,
}
impl Lifecycle {
    #[must_use]
    pub fn as_column(self) -> &'static str {
        match self {
            Lifecycle::Active => LIFECYCLE_ACTIVE,
            Lifecycle::Paused => LIFECYCLE_PAUSED,
            Lifecycle::Cancelling => LIFECYCLE_CANCELLING,
        }
    }
    #[must_use]
    pub fn from_column(column: &str) -> Option<Self> {
        match column {
            LIFECYCLE_ACTIVE => Some(Lifecycle::Active),
            LIFECYCLE_PAUSED => Some(Lifecycle::Paused),
            LIFECYCLE_CANCELLING => Some(Lifecycle::Cancelling),
            _ => None,
        }
    }
}
// JSON encodings of `PendingStateFinishedResultKind` as stored in the `result_kind` column,
// pinned by `result_kind_json_constants_match_serde`.
pub const RESULT_KIND_JSON_OK: &str = r#""ok""#;
pub const RESULT_KIND_JSON_ERROR: &str = r#"{"err":"error"}"#;
pub const HISTORY_EVENT_TYPE_JOIN_NEXT: &str = "join_next"; // Serialization tag of `HistoryEvent::JoinNext`

#[derive(Debug, PartialEq, Eq, Clone)]
pub struct ExecutionLog {
    pub execution_id: ExecutionId,
    pub events: Vec<ExecutionEvent>,
    pub responses: Vec<ResponseWithCursor>,
    pub next_version: Version, // Is not advanced once in Finished state
    pub pending_state: PendingState, // reflecting the current state
    pub component_digest: ComponentDigest, // reflecting the current state
    pub component_type: ComponentType,
    pub deployment_id: DeploymentId, // reflecting the current state
}

impl ExecutionLog {
    /// Return some duration after which the execution will be retried.
    /// Return `None` if no more retries are allowed.
    #[must_use]
    pub fn can_be_retried_after(
        temporary_event_count: u32,
        max_retries: Option<u32>,
        retry_exp_backoff: Duration,
    ) -> Option<Duration> {
        // If max_retries == None, wrapping is OK after this succeeds - we want to retry forever.
        if temporary_event_count <= max_retries.unwrap_or(u32::MAX) {
            // TODO: Add test for number of retries
            let duration = retry_exp_backoff * 2_u32.saturating_pow(temporary_event_count - 1);
            Some(duration)
        } else {
            None
        }
    }

    #[must_use]
    pub fn compute_retry_duration_when_retrying_forever(
        temporary_event_count: u32,
        retry_exp_backoff: Duration,
    ) -> Duration {
        Self::can_be_retried_after(temporary_event_count, None, retry_exp_backoff)
            .expect("`max_retries` set to MAX must never return None")
    }

    #[must_use]
    pub fn get_create_request(&self) -> CreateRequest {
        assert_matches!(self.events.first().cloned(), Some(ExecutionEvent {
            event:ExecutionRequest::Created{
                ffqn,params,parent,scheduled_at,component_id,deployment_id,metadata,scheduled_by},
                created_at, .. }) => CreateRequest { created_at, execution_id:
                    self.execution_id.clone(), ffqn, params, parent, scheduled_at,
                    component_id, deployment_id, metadata, scheduled_by, paused: false })
    }

    #[must_use]
    pub fn ffqn(&self) -> &FunctionFqn {
        assert_matches!(self.events.first(), Some(ExecutionEvent {
            event: ExecutionRequest::Created { ffqn, .. },
            ..
        }) => ffqn)
    }

    #[must_use]
    pub fn params(&self) -> &Params {
        assert_matches!(self.events.first(), Some(ExecutionEvent {
            event: ExecutionRequest::Created { params, .. },
            ..
        }) => params)
    }

    #[must_use]
    pub fn parent(&self) -> Option<(ExecutionId, JoinSetId)> {
        assert_matches!(self.events.first(), Some(ExecutionEvent {
            event: ExecutionRequest::Created { parent, .. },
            ..
        }) => parent.clone())
    }

    #[must_use]
    pub fn last_event(&self) -> &ExecutionEvent {
        self.events.last().expect("must contain at least one event")
    }

    #[must_use]
    pub fn is_finished(&self) -> bool {
        matches!(
            self.events.last(),
            Some(ExecutionEvent {
                event: ExecutionRequest::Finished { .. },
                ..
            })
        )
    }

    #[must_use]
    pub fn as_finished_result(&self) -> Option<SupportedFunctionReturnValue> {
        if let ExecutionEvent {
            event: ExecutionRequest::Finished { retval: result, .. },
            ..
        } = self.events.last().expect("must contain at least one event")
        {
            Some(result.clone())
        } else {
            None
        }
    }

    pub fn event_history(&self) -> impl Iterator<Item = (HistoryEvent, Version)> + '_ {
        self.events.iter().filter_map(|event| {
            if let ExecutionRequest::HistoryEvent { event: eh, .. } = &event.event {
                Some((eh.clone(), event.version.clone()))
            } else {
                None
            }
        })
    }

    #[cfg(feature = "test")]
    #[must_use]
    pub fn find_join_set_request(&self, join_set_id: &JoinSetId) -> Option<&JoinSetRequest> {
        self.events
            .iter()
            .find_map(move |event| match &event.event {
                ExecutionRequest::HistoryEvent {
                    event:
                        HistoryEvent::JoinSetRequest {
                            join_set_id: found,
                            request,
                        },
                    ..
                } if *join_set_id == *found => Some(request),
                _ => None,
            })
    }
}

pub type VersionType = u32;
#[derive(
    Debug,
    Default,
    Clone,
    PartialEq,
    PartialOrd,
    Ord,
    Eq,
    Hash,
    derive_more::Display,
    derive_more::Into,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
)]
#[serde(transparent)]
#[schemars(transparent)]
pub struct Version(pub VersionType);
impl Version {
    #[must_use]
    pub fn new(arg: VersionType) -> Version {
        Version(arg)
    }

    #[must_use]
    pub fn increment(&self) -> Version {
        Version(self.0 + 1)
    }
}
impl TryFrom<i64> for Version {
    type Error = VersionParseError;
    fn try_from(value: i64) -> Result<Self, Self::Error> {
        VersionType::try_from(value)
            .map(Version::new)
            .map_err(|_| VersionParseError)
    }
}
impl From<Version> for usize {
    fn from(value: Version) -> Self {
        usize::try_from(value.0).expect("16 bit systems are unsupported")
    }
}
impl From<&Version> for usize {
    fn from(value: &Version) -> Self {
        usize::try_from(value.0).expect("16 bit systems are unsupported")
    }
}

#[derive(Debug, thiserror::Error)]
#[error("version must be u32")]
pub struct VersionParseError;

#[derive(
    Clone,
    Debug,
    derive_more::Display,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
)]
#[display("{event}")]
pub struct ExecutionEvent {
    pub created_at: DateTime<Utc>,
    pub event: ExecutionRequest,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub backtrace_id: Option<Version>,
    pub version: Version,
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    derive_more::Display,
    derive_more::Into,
    Serialize,
    /* webapi */ Deserialize,
    schemars::JsonSchema,
)]
pub struct ResponseCursor(pub u32);

#[derive(
    Debug, Clone, PartialEq, Eq, Serialize /* webapi */, Deserialize, schemars::JsonSchema,
)]
pub struct ResponseWithCursor {
    pub event: JoinSetResponseEventOuter,
    pub cursor: ResponseCursor,
}

#[derive(Debug)]
pub struct ListExecutionEventsResponse {
    pub events: Vec<ExecutionEvent>,
    pub max_version: Version,
}

#[derive(Debug)]
pub struct ListResponsesResponse {
    pub responses: Vec<ResponseWithCursor>,
    pub max_cursor: ResponseCursor,
    pub scan_cursor: ResponseCursor,
}

impl ListResponsesResponse {
    #[must_use]
    pub fn from_page(
        responses: Vec<ResponseWithCursor>,
        max_cursor: ResponseCursor,
        pagination: Pagination<u32>,
    ) -> Self {
        let scan_cursor = match pagination {
            Pagination::NewerThan { length, .. } => match responses.last() {
                Some(response) if responses.len() >= usize::from(length.get()) => response.cursor,
                _ => max_cursor,
            },
            Pagination::OlderThan { .. } => responses
                .first()
                .map(|response| response.cursor)
                .unwrap_or(ResponseCursor(0)),
        };
        Self {
            responses,
            max_cursor,
            scan_cursor,
        }
    }
}

#[derive(
    Debug, Clone, PartialEq, Eq, Serialize /* webapi */, Deserialize, schemars::JsonSchema,
)]
pub struct JoinSetResponseEventOuter {
    pub created_at: DateTime<Utc>,
    pub event: JoinSetResponseEvent,
}

#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
pub struct JoinSetResponseEvent {
    pub join_set_id: JoinSetId,
    pub event: JoinSetResponse,
}

#[derive(
    Clone, Debug, PartialEq, Eq, Serialize, Deserialize, derive_more::Display, schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum JoinSetResponse {
    #[display("delay {}: {delay_id}", if result.is_ok() { "finished" } else { "cancelled"})]
    DelayFinished {
        delay_id: DelayId,
        result: Result<(), ()>,
    },
    #[display("{result}: {child_execution_id}")] // execution completed..
    ChildExecutionFinished {
        child_execution_id: ExecutionIdDerived,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Version(2)))]
        finished_version: Version,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = crate::SUPPORTED_RETURN_VALUE_OK_EMPTY))]
        result: SupportedFunctionReturnValue,
    },
}

pub const DUMMY_CREATED: ExecutionRequest = ExecutionRequest::Created {
    ffqn: FunctionFqn::new_static("", ""),
    params: Params::empty(),
    parent: None,
    scheduled_at: DateTime::from_timestamp_nanos(0),
    component_id: ComponentId::dummy_activity(),
    deployment_id: DeploymentId::from_parts(0, 0),
    metadata: ExecutionMetadata::empty(),
    scheduled_by: None,
};
pub const DUMMY_HISTORY_EVENT: ExecutionRequest = ExecutionRequest::HistoryEvent {
    event: HistoryEvent::JoinSetCreate {
        join_set_id: JoinSetId {
            kind: crate::JoinSetKind::OneOff,
            name: StrVariant::empty(),
        },
    },
};

#[derive(
    Clone,
    derive_more::Debug,
    derive_more::Display,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(rename_all = "snake_case")]
pub enum ExecutionRequest {
    #[display("Created({ffqn}, `{scheduled_at}`)")]
    Created {
        ffqn: FunctionFqn,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Params::empty()))]
        #[debug(skip)]
        params: Params,
        parent: Option<(ExecutionId, JoinSetId)>,
        scheduled_at: DateTime<Utc>,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentId::dummy_activity()))]
        component_id: ComponentId,
        deployment_id: DeploymentId,
        #[cfg_attr(any(test, feature = "test"), arbitrary(default))]
        metadata: ExecutionMetadata,
        scheduled_by: Option<ExecutionId>,
    },
    Locked(Locked),
    /// Releases a lock.
    ///
    /// State transition semantics:
    /// - [`PendingState::Locked`] becomes [`PendingState::PendingAt`] at
    ///   [`Unlocked::backoff_expires_at`]. The field name is kept for persisted JSON and gRPC
    ///   compatibility, but it is the next pending instant for every unlock reason.
    /// - [`PendingState::PendingAt`], [`PendingState::BlockedByJoinSet`],
    ///   [`PendingState::Paused`], and [`PendingState::Finished`] reject this event.
    #[display("Unlocked({_0})")]
    Unlocked(Unlocked),
    /// Does not change `PendingState`.
    #[display("ComponentUpgradeFinished({component_digest})")]
    ComponentUpgradeFinished {
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentId::dummy_activity().component_digest))]
        component_digest: ComponentDigest,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = DeploymentId::from_parts(0, 0)))]
        deployment_id: DeploymentId,
        outcome: ComponentUpgradeOutcome,
    },
    // Created by the executor holding the lock.
    // After expiry interpreted as pending.
    #[display("TemporarilyFailed(`{backoff_expires_at}`)")]
    TemporarilyFailed {
        backoff_expires_at: DateTime<Utc>,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
        reason: StrVariant,
        detail: Option<String>,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = None))]
        http_client_traces: Option<Vec<HttpClientTrace>>,
    },
    // Created by the executor holding the lock.
    // After expiry interpreted as pending.
    #[display("TemporarilyTimedOut(`{backoff_expires_at}`)")]
    TemporarilyTimedOut {
        backoff_expires_at: DateTime<Utc>,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = None))]
        http_client_traces: Option<Vec<HttpClientTrace>>,
    },
    // Created by the executor holding the lock.
    #[display("Finished: {retval}")]
    Finished {
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = crate::SUPPORTED_RETURN_VALUE_OK_EMPTY))]
        retval: SupportedFunctionReturnValue,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = None))]
        http_client_traces: Option<Vec<HttpClientTrace>>,
    },

    #[display("HistoryEvent({event})")]
    HistoryEvent {
        event: HistoryEvent,
    },
    #[display("Paused")]
    Paused,
    #[display("Unpaused")]
    Unpaused,
    /// Requests cancellation. Sets `lifecycle` to `cancelling` without changing
    /// the underlying state; the cancellation driver or activity owner then
    /// appends `Finished(Cancelled)`.
    ///
    /// State transition semantics (mirrors [`ExecutionRequest::Paused`]):
    /// - Every non-terminal state sets `lifecycle = cancelling`, underlying state
    ///   unchanged. A paused execution is never running; a locked activity keeps
    ///   its lock until teardown or lease expiry; a locked workflow's run is
    ///   fenced by this event's version bump.
    /// - [`PendingState::Finished`] is rejected (already terminal).
    #[display("CancellationRequested")]
    CancellationRequested,
}

/// Reason for auditing only
#[derive(
    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ComponentUpgradeReason {
    #[display("auto")]
    Auto,
    #[display("manual(force = {force})")]
    Manual { force: bool },
}

#[derive(
    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[display("{reason}, pending at {unlocked_at}")]
pub struct Unlocked {
    /// Instant used when releasing a currently locked execution back to [`PendingState::PendingAt`].
    // The field was renamed only in Rust; keep its serialized name stable.
    #[serde(rename = "backoff_expires_at")]
    pub unlocked_at: DateTime<Utc>,
    #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
    pub reason: StrVariant,
}

#[derive(
    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ComponentUpgradeOutcome {
    #[display("success({reason})")]
    Success { reason: ComponentUpgradeReason },
    #[display("failed: {reason}")]
    Failed {
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
        reason: StrVariant,
    },
}

impl ExecutionRequest {
    #[must_use]
    pub fn is_temporary_event(&self) -> bool {
        matches!(
            self,
            Self::TemporarilyFailed { .. } | Self::TemporarilyTimedOut { .. }
        )
    }

    /// String representation of `ExecutionRequest`, used in execution log table to fetch events of certain type, e.g. `created` + `history_event`.
    #[must_use]
    pub const fn variant(&self) -> &'static str {
        match self {
            ExecutionRequest::Created { .. } => "created",
            ExecutionRequest::Locked(_) => "locked",
            ExecutionRequest::Unlocked(_) => "unlocked",
            ExecutionRequest::ComponentUpgradeFinished { .. } => "component_upgrade_finished",
            ExecutionRequest::TemporarilyFailed { .. } => "temporarily_failed",
            ExecutionRequest::TemporarilyTimedOut { .. } => "temporarily_timed_out",
            ExecutionRequest::Finished { .. } => "finished",
            ExecutionRequest::HistoryEvent { .. } => "history_event",
            ExecutionRequest::Paused => "paused",
            ExecutionRequest::Unpaused => "unpaused",
            ExecutionRequest::CancellationRequested => "cancellation_requested",
        }
    }

    #[must_use]
    pub fn join_set_id(&self) -> Option<&JoinSetId> {
        match self {
            Self::Created {
                parent: Some((_parent_id, join_set_id)),
                ..
            } => Some(join_set_id),
            Self::HistoryEvent {
                event:
                    HistoryEvent::JoinSetCreate { join_set_id, .. }
                    | HistoryEvent::JoinSetRequest { join_set_id, .. }
                    | HistoryEvent::JoinNext { join_set_id, .. },
            } => Some(join_set_id),
            _ => None,
        }
    }
}

#[derive(
    Clone,
    derive_more::Debug,
    derive_more::Display,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[display("Locked(`{lock_expires_at}`, {component_id})")]
pub struct Locked {
    #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentId::dummy_activity()))]
    pub component_id: ComponentId,
    pub executor_id: ExecutorId,
    pub deployment_id: DeploymentId,
    pub run_id: RunId,
    pub lock_expires_at: DateTime<Utc>,
    #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentRetryConfig::ZERO))]
    pub retry_config: ComponentRetryConfig,
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    derive_more::Display,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum PersistKind {
    #[display("RandomU64({min}, {max_inclusive})")]
    RandomU64 {
        min: u64,
        max_inclusive: u64,
    },
    #[display("RandomString({min_length}, {max_length_exclusive})")]
    RandomString {
        min_length: u64,
        max_length_exclusive: u64,
    },
    ExecutionId,
}

#[must_use]
pub fn from_u64_to_bytes(value: u64) -> [u8; 8] {
    value.to_be_bytes()
}

#[derive(
    derive_more::Debug,
    Clone,
    PartialEq,
    Eq,
    derive_more::Display,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
/// Must be created by the executor in [`PendingState::Locked`].
pub enum HistoryEvent {
    /// Persist a generated pseudorandom value.
    #[display("Persist")]
    Persist {
        #[debug(skip)]
        value: Vec<u8>, // Only stored for nondeterminism checks. TODO: Consider using a hashed value or just the intention.
        kind: PersistKind,
    },
    #[display("JoinSetCreate({join_set_id})")]
    JoinSetCreate { join_set_id: JoinSetId },
    #[display("JoinSetRequest({request})")]
    // join_set_id is part of ExecutionId or DelayId in the `request`
    JoinSetRequest {
        join_set_id: JoinSetId,
        request: JoinSetRequest,
    },
    /// Sets the pending state to [`PendingState::BlockedByJoinSet`].
    /// When the response arrives at `resp_time`:
    /// The execution is [`PendingState::PendingAt`]`(max(resp_time, lock_expires_at)`, so that the
    /// original executor can continue. After the expiry any executor can continue without
    /// marking the execution as timed out.
    #[display("JoinNext({join_set_id})")]
    JoinNext {
        join_set_id: JoinSetId,
        /// Set to a future time if the worker is keeping the execution invocation warm waiting for the result.
        /// The pending status will be kept in Locked state until `run_expires_at`.
        run_expires_at: DateTime<Utc>,
        /// Set to a specific function when calling `-await-next` extension function, used for
        /// determinism checks.
        requested_ffqn: Option<FunctionFqn>,
        /// Closing request must never set `requested_ffqn` and is ignored by determinism checks.
        closing: bool,
    },
    /// Attempt to process next response without changing the pending state.
    #[display("JoinNextTry({join_set_id}, {outcome})")]
    JoinNextTry {
        join_set_id: JoinSetId,
        outcome: JoinNextTryOutcome,
    },
    /// Records the fact that a join set was awaited more times than its submission count.
    #[display("JoinNextTooMany({join_set_id})")]
    JoinNextTooMany {
        join_set_id: JoinSetId,
        /// Set to a specific function when calling `-await-next` extension function, used for
        /// determinism checks.
        requested_ffqn: Option<FunctionFqn>,
    },
    #[display("Schedule({execution_id}, {schedule_at})")]
    Schedule {
        execution_id: ExecutionId,
        schedule_at: HistoryEventScheduleAt, // Stores intention to schedule an execution at a date/offset
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Ok(())))]
        result: Result<(), ScheduleRequestError>,
    },
    #[display("Stub({target_execution_id})")]
    Stub {
        target_execution_id: ExecutionIdDerived,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StubRetVal::Typed(crate::SUPPORTED_RETURN_VALUE_OK_EMPTY).hash()))]
        retval_hash: StubRetValHash,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Ok(())))]
        result: Result<(), StubError>,
    },
}

/// Stub return value - only used during processing, not stored in history.
#[derive(derive_more::Debug, Clone, PartialEq, Eq)]
#[cfg_attr(any(test, feature = "test"), derive(Serialize, Deserialize))]
#[cfg_attr(any(test, feature = "test"), serde(rename_all = "snake_case"))]
pub enum StubRetVal {
    Typed(SupportedFunctionReturnValue),
    Untyped(String),
}

impl StubRetVal {
    /// Compute a stable hash of the return value for determinism checks.
    #[must_use]
    pub fn hash(&self) -> StubRetValHash {
        use sha2::{Digest as _, Sha256};
        const STUB_RETVAL_HASH_VERSION: u8 = 1;
        let mut hasher = Sha256::default();

        match self {
            StubRetVal::Typed(val) => {
                hasher.update(b"T|");
                // Serialize to JSON for stable hashing
                let json = serde_json::to_string(val)
                    .expect("SupportedFunctionReturnValue is always serializable");
                hasher.update(json.as_bytes());
            }
            StubRetVal::Untyped(s) => {
                hasher.update(b"U|");
                hasher.update(s.as_bytes());
            }
        }

        let hash_bytes = hasher.finalize();
        let mut result = [0u8; 33];
        result[0] = STUB_RETVAL_HASH_VERSION;
        result[1..].copy_from_slice(&hash_bytes);

        StubRetValHash(result)
    }
}

/// Hash of a stub return value, stored in history for determinism checks.
/// Format: 1 byte version + 32 bytes SHA-256 hash.
#[derive(
    Clone,
    PartialEq,
    Eq,
    serde_with::SerializeDisplay,
    serde_with::DeserializeFromStr,
    schemars::JsonSchema,
)]
#[schemars(with = "String")]
pub struct StubRetValHash([u8; 33]);

impl Display for StubRetValHash {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        for b in self.0 {
            write!(f, "{b:02x}")?;
        }
        Ok(())
    }
}

impl Debug for StubRetValHash {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        Display::fmt(self, f)
    }
}

impl std::str::FromStr for StubRetValHash {
    type Err = StubRetValHashParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.len() != 66 {
            // 33 bytes * 2 hex chars = 66
            return Err(StubRetValHashParseError::InvalidLength(s.len()));
        }
        let mut bytes = [0u8; 33];
        for i in 0..33 {
            let chunk = &s[i * 2..i * 2 + 2];
            bytes[i] =
                u8::from_str_radix(chunk, 16).map_err(|_| StubRetValHashParseError::InvalidHex)?;
        }
        Ok(StubRetValHash(bytes))
    }
}

#[derive(Debug, thiserror::Error)]
pub enum StubRetValHashParseError {
    #[error("invalid length: expected 66 hex chars, got {0}")]
    InvalidLength(usize),
    #[error("invalid hex character")]
    InvalidHex,
}

/// Error from the `-stub` extension function.
/// Mirrors `obelisk:types/execution.{stub-error}` from WIT.
#[derive(
    Debug, Clone, thiserror::Error, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum StubError {
    #[error("execution not found")]
    ExecutionNotFound,
    #[error("type check error: {0}")]
    TypeCheckError(String),
    #[error("conflict")]
    Conflict,
}

/// Error from the `schedule-json` function. Persisted in history for determinism.
#[derive(
    Debug, Clone, thiserror::Error, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ScheduleRequestError {
    #[error("function not found")]
    FunctionNotFound,
    #[error("params parsing error: {0}")]
    TypeCheckError(String),
}

/// Error from the `submit-json` function. Persisted in history for determinism.
#[derive(
    Debug, Clone, thiserror::Error, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ChildExecutionRequestError {
    #[error("function not found")]
    FunctionNotFound,
    #[error("params parsing error: {0}")]
    TypeCheckError(String),
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    derive_more::Display,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(rename_all = "snake_case")]
pub enum JoinNextTryOutcome {
    /// A response was found and processed.
    #[display("found")]
    Found,
    /// No response available, but there are still pending requests.
    #[display("pending")]
    Pending,
    /// No response available, and all requests have been processed.
    #[display("all_processed")]
    AllProcessed,
}

impl From<bool> for JoinNextTryOutcome {
    /// Migration helper: converts old `found_response: bool` to the new enum.
    /// `false` maps to `Pending` as a conservative default (the exact error
    /// was not stored before).
    fn from(found_response: bool) -> Self {
        if found_response {
            JoinNextTryOutcome::Found
        } else {
            JoinNextTryOutcome::Pending
        }
    }
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    derive_more::Display,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(rename_all = "snake_case")]
pub enum HistoryEventScheduleAt {
    Now,
    #[display("At(`{_0}`)")]
    At(DateTime<Utc>),
    #[display("In({_0:?})")]
    In(Duration),
}

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ScheduleAtConversionError {
    #[error("source duration value is out of range")]
    OutOfRangeError,
}

impl HistoryEventScheduleAt {
    pub fn as_date_time(
        &self,
        now: DateTime<Utc>,
    ) -> Result<DateTime<Utc>, ScheduleAtConversionError> {
        match self {
            Self::Now => Ok(now),
            Self::At(date_time) => Ok(*date_time),
            Self::In(duration) => {
                let time_delta = TimeDelta::from_std(*duration)
                    .map_err(|_| ScheduleAtConversionError::OutOfRangeError)?;
                now.checked_add_signed(time_delta)
                    .ok_or(ScheduleAtConversionError::OutOfRangeError)
            }
        }
    }
}

#[derive(
    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum JoinSetRequest {
    // Must be created by the executor in `PendingState::Locked`.
    #[display("DelayRequest({delay_id}, expires_at: `{expires_at}`, schedule_at: `{schedule_at}`)")]
    DelayRequest {
        delay_id: DelayId,
        expires_at: DateTime<Utc>,
        schedule_at: HistoryEventScheduleAt,
        #[serde(default)]
        paused: bool,
    },
    // Must be created by the executor in `PendingState::Locked`.
    #[display("ChildExecutionRequest({child_execution_id}, {target_ffqn}, params: {params})")]
    ChildExecutionRequest {
        child_execution_id: ExecutionIdDerived,
        target_ffqn: FunctionFqn,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Params::empty()))]
        params: Params,
        #[cfg_attr(any(test, feature = "test"), arbitrary(value = Ok(())))]
        result: Result<(), ChildExecutionRequestError>,
    },
}

/// Error that is not specific to an execution.
#[derive(Debug, Clone, thiserror::Error, derive_more::PartialEq, derive_more::Eq)]
pub enum DbErrorGeneric {
    #[error("database error: {reason}")]
    Uncategorized {
        reason: StrVariant,
        #[eq(skip)]
        #[partial_eq(skip)]
        context: SpanTrace,
        #[eq(skip)]
        #[partial_eq(skip)]
        #[source]
        source: Option<Arc<dyn std::error::Error + Send + Sync>>,
        loc: &'static Location<'static>,
    },
    #[error("database was closed")]
    Close,
}

#[derive(thiserror::Error, Clone, Debug, derive_more::PartialEq, derive_more::Eq)]
pub enum DbErrorWriteNonRetriable {
    #[error("validation failed: {0}")]
    ValidationFailed(StrVariant),
    #[error("conflict")]
    Conflict,
    #[error("already finished")]
    AlreadyFinished,
    #[error("illegal state: {reason}")]
    IllegalState {
        reason: StrVariant,
        #[eq(skip)]
        #[partial_eq(skip)]
        context: SpanTrace,
        #[eq(skip)]
        #[partial_eq(skip)]
        #[source]
        source: Option<Arc<dyn std::error::Error + Send + Sync>>,
        loc: &'static Location<'static>,
    },
    #[error("illegal state: `Unlocked` cannot be appended in state {0}")]
    UnlockedCannotBeAppended(&'static str),
    #[error("version conflict: expected: {expected}, got: {requested}")]
    VersionConflict {
        expected: Version,
        requested: Version,
    },
}

/// Write error tied to an execution
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
pub enum DbErrorWrite {
    #[error("cannot write - row not found")]
    NotFound,
    #[error("non-retriable error: {0}")]
    NonRetriable(#[from] DbErrorWriteNonRetriable),
    #[error(transparent)]
    Generic(#[from] DbErrorGeneric),
}

/// Error from idempotent stub response write.
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
pub enum DbErrorStubResponse {
    #[error("stub conflict: already finished with a different value")]
    StubConflict,
    #[error(transparent)]
    Write(#[from] DbErrorWrite),
}

/// Read error tied to an execution
#[derive(Debug, Clone, thiserror::Error, PartialEq)]
pub enum DbErrorRead {
    #[error("cannot read - row not found")]
    NotFound,
    #[error(transparent)]
    Generic(#[from] DbErrorGeneric),
}

#[derive(Debug, thiserror::Error, PartialEq)]
pub enum DbErrorReadWithTimeout {
    #[error("timeout")]
    Timeout(TimeoutOutcome),
    #[error(transparent)]
    DbErrorRead(#[from] DbErrorRead),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ResponseSubscriptionEnd {
    PollIntervalElapsed,
    LockDeadlineReached,
    ExecutorClosing,
    ExecutionUpdated,
}

#[derive(Debug, thiserror::Error, PartialEq)]
pub enum SubscribeToResponsesError {
    #[error("response subscription ended: {0:?}")]
    SubscriptionEnded(ResponseSubscriptionEnd),
    #[error(transparent)]
    DbErrorRead(#[from] DbErrorRead),
}

// Represents next version after successfuly appended to execution log.
// TODO: Convert to struct with next_version
pub type AppendResponse = Version;
pub type PendingExecution = (ExecutionId, Version, Params, Option<DateTime<Utc>>);

#[derive(Debug, Clone)]
pub struct LockedExecution {
    pub execution_id: ExecutionId,
    pub next_version: Version,
    pub metadata: ExecutionMetadata,
    pub component_digest: ComponentDigest,
    pub locked_event: Locked,
    pub ffqn: FunctionFqn,
    pub params: Params,
    pub event_history: Vec<(HistoryEvent, Version)>,
    pub responses: Vec<ResponseWithCursor>,
    pub parent: Option<(ExecutionId, JoinSetId)>,
    pub intermittent_event_count: u32,
}

pub type LockPendingResponse = Vec<LockedExecution>;
pub type AppendBatchResponse = Version;

#[derive(Debug, Clone, PartialEq, derive_more::Display, Serialize, Deserialize)]
#[display("{event}")]
pub struct AppendRequest {
    pub created_at: DateTime<Utc>,
    pub event: ExecutionRequest,
}

#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "test", derive(Serialize))]
pub struct CreateRequest {
    pub created_at: DateTime<Utc>,
    pub execution_id: ExecutionId,
    pub ffqn: FunctionFqn,
    pub params: Params,
    pub parent: Option<(ExecutionId, JoinSetId)>,
    pub scheduled_at: DateTime<Utc>,
    pub component_id: ComponentId,
    pub deployment_id: DeploymentId,
    pub metadata: ExecutionMetadata,
    pub scheduled_by: Option<ExecutionId>,
    pub paused: bool,
}

impl From<CreateRequest> for ExecutionRequest {
    fn from(value: CreateRequest) -> Self {
        Self::Created {
            ffqn: value.ffqn,
            params: value.params,
            parent: value.parent,
            scheduled_at: value.scheduled_at,
            component_id: value.component_id,
            deployment_id: value.deployment_id,
            metadata: value.metadata,
            scheduled_by: value.scheduled_by,
        }
    }
}

#[async_trait]
pub trait DbPool: Send + Sync {
    async fn db_exec_conn(&self) -> Result<Box<dyn DbExecutor>, DbErrorGeneric>;

    async fn connection(&self) -> Result<Box<dyn DbConnection>, DbErrorGeneric>;

    async fn external_api_conn(&self) -> Result<Box<dyn DbExternalApi>, DbErrorGeneric>;

    /// Content-addressed blob store for deployment files. Separate from the metadata
    /// connections so the bytes can move to an object store (S3) in future while the
    /// referencing metadata stays in the database.
    async fn cas_conn(&self) -> Result<Box<dyn crate::cas::Cas>, DbErrorGeneric>;

    #[cfg(feature = "test")]
    async fn connection_test(&self) -> Result<Box<dyn DbConnectionTest>, DbErrorGeneric>;
}

#[async_trait]
pub trait DbPoolCloseable {
    async fn close(&self);
}

#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "test", derive(Serialize))]
pub struct AppendEventsToExecution {
    pub execution_id: ExecutionId,
    pub version: Version,
    pub batch: Vec<AppendRequest>,
}

#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "test", derive(Serialize))]
pub struct AppendResponseToExecution {
    pub parent_execution_id: ExecutionId,
    pub created_at: DateTime<Utc>,
    pub join_set_id: JoinSetId,
    pub child_execution_id: ExecutionIdDerived,
    pub finished_version: Version,
    pub result: SupportedFunctionReturnValue,
}

/// A captured database write operation with all arguments needed to replay it
/// against the real database.
/// Dates carry meaning only on a fresh replay, ignoring user's input when persisting.
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "test", derive(Serialize))]
pub enum CapturedDbWrite {
    Append {
        execution_id: ExecutionId,
        version: Version,
        req: AppendRequest,
        backtraces: Vec<BacktraceInfo>,
    },
    AppendBatch {
        current_time: DateTime<Utc>,
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
        backtraces: Vec<BacktraceInfo>,
    },
    AppendBatchWithDelayResponse {
        current_time: DateTime<Utc>,
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
        join_set_id: JoinSetId,
        delay_id: DelayId,
        backtraces: Vec<BacktraceInfo>,
    },
    AppendBatchCreateNewExecution {
        current_time: DateTime<Utc>,
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
        child_req: Vec<CreateRequest>,
        backtraces: Vec<BacktraceInfo>,
    },
    AppendStubResponse {
        events: AppendEventsToExecution,
        response: AppendResponseToExecution,
        current_time: DateTime<Utc>,
        backtraces: Vec<BacktraceInfo>,
    },
    AppendFinished {
        execution_id: ExecutionId,
        version: Version,
        current_time: DateTime<Utc>,
        retval: SupportedFunctionReturnValue,
        parent: Option<(ExecutionId, JoinSetId)>,
    },
}
impl CapturedDbWrite {
    #[must_use]
    pub fn is_finished(&self) -> bool {
        matches!(self, CapturedDbWrite::AppendFinished { .. })
    }
}

#[async_trait]
pub trait DbExecutor: Send + Sync {
    #[expect(clippy::too_many_arguments)]
    async fn lock_pending_by_ffqns(
        &self,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        ffqns: Arc<[FunctionFqn]>,
        created_at: DateTime<Utc>,
        component_id: ComponentId,
        deployment_id: DeploymentId,
        executor_id: ExecutorId,
        lock_expires_at: DateTime<Utc>,
        run_id: RunId,
        retry_config: ComponentRetryConfig,
    ) -> Result<LockPendingResponse, DbErrorWrite>;

    #[expect(clippy::too_many_arguments)]
    async fn lock_pending_by_ffqns_auto(
        &self,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        ffqns: Arc<[FunctionFqn]>,
        created_at: DateTime<Utc>,
        component_id: ComponentId,
        deployment_id: DeploymentId,
        executor_id: ExecutorId,
        lock_expires_at: DateTime<Utc>,
        run_id: RunId,
        retry_config: ComponentRetryConfig,
    ) -> Result<LockPendingResponse, DbErrorWrite>;

    #[expect(clippy::too_many_arguments)]
    async fn lock_pending_by_component_digest(
        &self,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        component_id: &ComponentId,
        deployment_id: DeploymentId,
        created_at: DateTime<Utc>,
        executor_id: ExecutorId,
        lock_expires_at: DateTime<Utc>,
        run_id: RunId,
        retry_config: ComponentRetryConfig,
    ) -> Result<LockPendingResponse, DbErrorWrite>;

    #[cfg(feature = "test")]
    #[expect(clippy::too_many_arguments)]
    async fn lock_one(
        &self,
        created_at: DateTime<Utc>,
        component_id: ComponentId,
        deployment_id: DeploymentId,
        execution_id: &ExecutionId,
        run_id: RunId,
        version: Version,
        executor_id: ExecutorId,
        lock_expires_at: DateTime<Utc>,
        retry_config: ComponentRetryConfig,
    ) -> Result<LockedExecution, DbErrorWrite>;

    /// Append a single event to an existing execution log.
    /// The request cannot contain [`ExecutionRequest::Created`].
    async fn append(
        &self,
        execution_id: ExecutionId,
        version: Version,
        req: AppendRequest,
    ) -> Result<AppendResponse, DbErrorWrite>;

    /// Append a batch of events to an existing execution log, and append a response to a parent execution.
    /// The batch cannot contain [`ExecutionRequest::Created`].
    async fn append_batch_respond_to_parent(
        &self,
        events: AppendEventsToExecution,
        response: AppendResponseToExecution,
        current_time: DateTime<Utc>, // not persisted, can be used for unblocking `subscribe_to_pending`
    ) -> Result<AppendBatchResponse, DbErrorWrite>;

    /// Notification mechainism with no strict guarantees for waiting while there are no pending executions.
    /// Return immediately if there are pending notifications at `pending_at_or_sooner`.
    /// Otherwise wait until `timeout_fut` resolves.
    /// Delay requests that expire between `pending_at_or_sooner` and timeout can be disregarded.
    /// If `current_digest` is set, ignore executions with incompatible digests.
    async fn wait_for_pending_by_ffqn(
        &self,
        pending_at_or_sooner: DateTime<Utc>,
        ffqns: Arc<[FunctionFqn]>,
        current_digest: Option<ComponentDigest>,
        timeout_fut: Pin<Box<dyn Future<Output = ()> + Send>>,
    );

    /// Notification mechainism with no strict guarantees for waiting while there are no pending executions.
    /// Return immediately if there are pending notifications at `pending_at_or_sooner`.
    /// Otherwise wait until `timeout_fut` resolves.
    /// Delay requests that expire between `pending_at_or_sooner` and timeout can be disregarded.
    async fn wait_for_pending_by_component_digest(
        &self,
        pending_at_or_sooner: DateTime<Utc>,
        component_digest: &ComponentDigest,
        timeout_fut: Pin<Box<dyn Future<Output = ()> + Send>>,
    );

    /// See [`Self::append_activity_cancellation_requested`].
    async fn cancel_activity_with_retries(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        let mut retries = 5;
        loop {
            match self
                .append_activity_cancellation_requested(execution_id, cancelled_at)
                .await
            {
                Err(DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::VersionConflict {
                    ..
                })) if retries > 0 => retries -= 1,
                res => return res,
            }
        }
    }

    /// Request cancellation of a cancellable workflow. In one version-guarded
    /// transaction, appends [`ExecutionRequest::CancellationRequested`]; rejects a non-cancellable
    /// target and returns `AlreadyFinished`/`AlreadyCancelling` without appending.
    /// The `Finished(Cancelled)` outcome is driven later by the cancellation driver.
    async fn cancel_workflow(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite>;

    /// Request cancellation of a cancellable workflow, retrying the version-guarded
    /// transaction on the live-worker race.
    async fn cancel_workflow_with_retries(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        let mut retries = 5;
        loop {
            match self.cancel_workflow(execution_id, cancelled_at).await {
                Err(DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::VersionConflict {
                    ..
                })) if retries > 0 => retries -= 1,
                res => return res,
            }
        }
    }

    /// Get last event. Impls may set `ExecutionEvent::backtrace_id` to `None`.
    async fn get_last_execution_event(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<ExecutionEvent, DbErrorRead>;

    /// Append [`ExecutionRequest::CancellationRequested`] if execution is not finished and not in
    /// cancellation already.
    /// The state will become [`PendingStateCancelling`] with the underlying state embedded.
    async fn append_activity_cancellation_requested(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite>;
}

pub enum AppendDelayResponseOutcome {
    Success,
    AlreadyFinished,
    AlreadyCancelled,
}

#[derive(Debug, Clone, Default)]
pub struct ListExecutionsFilter {
    pub function_name_filter: Option<FunctionNameFilter>,
    pub show_derived: bool,
    pub hide_finished: bool,
    pub execution_id_prefix: Option<String>,
    pub component_digest: Option<ComponentDigest>,
    pub deployment_id: Option<DeploymentId>,
    /// Match executions in any of the given states (logical OR). Empty list matches all.
    /// All [`ExecutionStateFilter::Pending`] / [`ExecutionStateFilter::Scheduled`] entries
    /// must carry the same `now`.
    pub state_filters: Vec<ExecutionStateFilter>,
}

/// Filter executions by their current state, using the same buckets as the deployment summary.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutionStateFilter {
    Locked,
    /// [`PendingState::PendingAt`] with the scheduled time at or before `now`.
    Pending {
        now: DateTime<Utc>,
    },
    /// [`PendingState::PendingAt`] with the scheduled time after `now`.
    Scheduled {
        now: DateTime<Utc>,
    },
    Blocked,
    /// Paused regardless of the underlying state (locked, pending or blocked).
    Paused,
    /// Cancellation requested; teardown in progress (underlying state pending or blocked).
    Cancelling,
    /// Finished with any result.
    Finished,
    /// Finished successfully.
    FinishedOk,
    /// Finished with the `err` variant of the result type.
    FinishedError,
    /// Execution failure: trap, timeout, nondeterminism, cancellation etc.
    FinishedExecutionFailure,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FunctionNameFilter {
    PackageName(String),
    InterfaceName(String),
    FunctionName(String),
}

impl FunctionNameFilter {
    #[must_use]
    pub fn like_pattern(&self) -> String {
        match self {
            Self::FunctionName(function_name) | Self::InterfaceName(function_name) => {
                format!("{function_name}%")
            }
            Self::PackageName(package_name) => {
                if let Some((pkg_fqn_without_version, version)) = package_name.rsplit_once('@')
                    && !version.is_empty()
                    && pkg_fqn_without_version.contains(':')
                {
                    format!("{pkg_fqn_without_version}/%@{version}.%")
                } else {
                    format!("{package_name}%")
                }
            }
        }
    }
}

#[async_trait]
pub trait DbExternalApi: DbConnection {
    /// Get the latest backtrace if version is not set.
    async fn get_backtrace(
        &self,
        execution_id: &ExecutionId,
        filter: BacktraceFilter,
    ) -> Result<BacktraceInfo, DbErrorRead>;

    /// Map a backtrace source file (a blob already stored in the CAS) to a component digest.
    /// `frame_key` is either an exact frame symbol path or a suffix (with leading `/`)
    /// when `is_suffix` is true. Repeated calls replace the digest mapped to the same key.
    /// The blob bytes themselves live in the CAS (see [`crate::cas::Cas`]); only this mapping
    /// lives in the database.
    async fn upsert_source_mapping(
        &self,
        component_digest: &ComponentDigest,
        frame_key: &str,
        is_suffix: bool,
        digest: &ContentDigest,
    ) -> Result<(), DbErrorWrite>;

    /// Resolve a backtrace source file's CAS digest by component digest and a frame symbol path.
    /// Matches either exact keys or suffix keys (where the frame path ends with the stored key).
    /// Returns `None` if not found or if multiple suffix entries match (ambiguous). The caller
    /// fetches the bytes from the CAS (see [`crate::cas::Cas`]).
    async fn resolve_source_digest(
        &self,
        component_digest: &ComponentDigest,
        file: &str,
    ) -> Result<Option<ContentDigest>, DbErrorRead>;

    /// Insert or reuse normalized component metadata rows.
    async fn upsert_component_metadata(
        &self,
        records: Vec<ComponentMetadataRecord>,
    ) -> Result<(), DbErrorWrite>;

    /// Insert deployment-local component bindings for a deployment.
    async fn insert_deployment_components(
        &self,
        deployment_id: DeploymentId,
        records: Vec<DeploymentComponentRecord>,
    ) -> Result<(), DbErrorWrite>;

    /// List all components visible in a deployment, including persisted imports, exports and WIT.
    async fn list_deployment_components(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<DeploymentComponentDetail>, DbErrorRead>;

    /// Get the WIT for a component digest scoped to a deployment.
    async fn get_deployment_component_wit(
        &self,
        deployment_id: DeploymentId,
        component_digest: &ComponentDigest,
    ) -> Result<Option<String>, DbErrorRead>;

    /// Returns executions sorted in descending order.
    async fn list_executions(
        &self,
        filter: ListExecutionsFilter,
        pagination: ExecutionListPagination,
    ) -> Result<Vec<ExecutionWithState>, DbErrorGeneric>;

    /// Returns execution events for the given execution.
    ///
    /// Results are always ordered from oldest to newest (ascending by version),
    /// regardless of pagination direction.
    async fn list_execution_events(
        &self,
        execution_id: &ExecutionId,
        pagination: Pagination<VersionType>,
        include_backtrace_id: bool,
    ) -> Result<ListExecutionEventsResponse, DbErrorRead>;

    /// Returns responses of an execution ordered as they arrived,
    /// enabling matching each `JoinNext` to its corresponding response.
    ///
    /// Results are always ordered from oldest to newest (ascending by cursor),
    /// regardless of pagination direction.
    ///
    /// As an optimization, the implementation can return an empty list of `responses`
    /// and `max_cursor` set to 0 if the execution is not found.
    async fn list_responses(
        &self,
        execution_id: &ExecutionId,
        pagination: Pagination<u32>,
    ) -> Result<ListResponsesResponse, DbErrorRead> {
        self.list_responses_filtered(execution_id, pagination, None)
            .await
    }

    async fn list_responses_filtered(
        &self,
        execution_id: &ExecutionId,
        pagination: Pagination<u32>,
        join_set: Option<&JoinSetId>,
    ) -> Result<ListResponsesResponse, DbErrorRead>;

    async fn list_execution_events_responses(
        &self,
        execution_id: &ExecutionId,
        req_since: &Version,
        req_max_length: NonZeroU16,
        req_include_backtrace_id: bool,
        resp_pagination: Pagination<VersionType>,
    ) -> Result<ExecutionWithStateRequestsResponses, DbErrorRead>;

    async fn upgrade_execution_component(
        &self,
        execution_id: &ExecutionId,
        old: &ComponentDigest,
        new: &ComponentDigest,
        reason: ComponentUpgradeReason,
    ) -> Result<(), DbErrorWrite>;

    async fn list_logs(
        &self,
        execution_id: &ExecutionId,
        show_derived: bool,
        filter: LogFilter,
        pagination: Pagination<LogCursor>,
    ) -> Result<ListLogsResponse, DbErrorRead>;

    async fn list_deployment_states(
        &self,
        current_time: DateTime<Utc>,
        pagination: Pagination<Option<DeploymentId>>,
        include_deployment_toml: bool,
        execution_counts: DeploymentExecutionCounts,
    ) -> Result<Vec<DeploymentState>, DbErrorRead>;

    /// Insert a new deployment row together with its component metadata and component rows in a
    /// single transaction, so the deployment row and its component rows commit together or not at
    /// all. The record must have `status == Inactive` and `last_active_at == None`; activation is
    /// a separate step via [`Self::activate_deployment`].
    async fn insert_deployment_with_components(
        &self,
        record: DeploymentRecord,
        component_metadata: Vec<ComponentMetadataRecord>,
        deployment_components: Vec<DeploymentComponentRecord>,
        deployment_component_files: Vec<DeploymentComponentFileRecord>,
    ) -> Result<(), DbErrorWrite>;

    /// Return deployment file digests referenced by this deployment but absent from the CAS.
    ///
    /// Blob bytes themselves are stored and fetched through the separate [`crate::cas::Cas`]
    /// trait (see [`DbPool::cas_conn`]); only this metadata/completeness query lives on the `Db`.
    async fn missing_digests(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<ContentDigest>, DbErrorRead>;

    /// Return the deployment-owned file refs recorded for a deployment.
    async fn list_deployment_files(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<DeploymentFileRecord>, DbErrorRead>;

    /// Delete content-addressed file blobs not referenced by any stored deployment,
    /// returning the number deleted. Such orphans are left behind when a submit writes
    /// blobs to the store and then fails verification before persisting the deployment.
    async fn gc_orphan_files(&self) -> Result<u64, DbErrorWrite>;

    async fn activate_deployment(
        &self,
        deployment_id: DeploymentId,
        now: DateTime<Utc>,
    ) -> Result<(), DbErrorWrite>;

    /// Mark a deployment as Enqueued (pending next server restart).
    /// Any previously Enqueued deployment is demoted to Inactive. If the target deployment is
    /// currently Active, it remains Active and any previously Enqueued deployment is cleared.
    /// The returned [`EnqueueOutcome`] reflects which of those happened.
    async fn enqueue_deployment(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<EnqueueOutcome, DbErrorWrite>;

    /// Returned [`DeploymentRecord`] must contain `deployment_toml`.
    async fn get_deployment(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Option<DeploymentRecord>, DbErrorRead>;

    /// Return active deployment.
    /// Returned [`DeploymentRecord`] must contain `deployment_toml`.
    #[cfg(feature = "test")]
    async fn get_active_deployment(&self) -> Result<Option<DeploymentRecord>, DbErrorRead>;

    /// Return the most relevant current deployment: Enqueued if present, otherwise Active.
    /// Returned [`DeploymentRecord`] must contain `deployment_toml`.
    async fn get_current_deployment(&self) -> Result<Option<DeploymentRecord>, DbErrorRead>;

    async fn list_deployments(
        &self,
        pagination: Pagination<Option<DeploymentId>>,
    ) -> Result<Vec<DeploymentRecord>, DbErrorRead>;

    /// Pause an execution.
    /// If the execution is an activity and is currently in `PendingState::Locked`, implementations must
    /// reject the write, otherwise a running activity will be considered terminated, which can break
    /// structured concurrency guarantees.
    async fn pause_execution(
        &self,
        execution_id: &ExecutionId,
        paused_at: DateTime<Utc>,
    ) -> Result<AppendResponse, DbErrorWrite>;

    /// Unpause an execution. Only paused executions can be unpaused.
    async fn unpause_execution(
        &self,
        execution_id: &ExecutionId,
        unpaused_at: DateTime<Utc>,
    ) -> Result<AppendResponse, DbErrorWrite>;

    /// Pause a delay, preventing it from being picked up by the expired timers watcher.
    /// No-op if the delay is already paused.
    /// Returns `NotFound` if the delay does not exist (already processed or cancelled).
    async fn pause_delay(&self, delay_id: &DelayId) -> Result<(), DbErrorWrite>;

    /// Unpause a previously paused delay.
    /// No-op if the delay is already unpaused.
    /// Returns `NotFound` if the delay does not exist (already processed or cancelled).
    async fn unpause_delay(&self, delay_id: &DelayId) -> Result<(), DbErrorWrite>;
}
pub const LIST_DEPLOYMENT_STATES_DEFAULT_LENGTH: u16 = 20;
pub const LIST_DEPLOYMENT_STATES_DEFAULT_PAGINATION: Pagination<Option<DeploymentId>> =
    Pagination::OlderThan {
        length: NonZeroU16::new(LIST_DEPLOYMENT_STATES_DEFAULT_LENGTH).unwrap(),
        cursor: None,
        including_cursor: false,
    };

pub struct DeploymentState {
    pub deployment_id: DeploymentId,
    pub description: Option<String>,
    /// Content digest = `sha256(deployment_toml)`.
    pub digest: ContentDigest,
    pub locked: u32,
    // In `PendingAt` state, scheduled to present or past
    pub pending: u32,
    // In `PendingAt` state, scheduled into the future
    pub scheduled: u32,
    pub blocked: u32,
    // Paused regardless of the underlying state (locked, pending or blocked).
    pub paused: u32,
    // Cancellation requested; teardown in progress. Disjoint from the buckets above.
    pub cancelling: u32,
    pub finished_ok: u32,
    pub finished_error: u32,
    pub finished_execution_failure: u32,
    /// Verbatim deployment manifest. None if not requested from db.
    pub deployment_toml: Option<String>,
    pub created_at: DateTime<Utc>,
    /// Set when the deployment becomes Active; None if it has never been active.
    pub last_active_at: Option<DateTime<Utc>>,
    pub status: DeploymentStatus,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeploymentExecutionCounts {
    /// Skip the aggregate queries; every bucket is reported as zero.
    Skip,
    /// Count executions per bucket; `include_derived` also counts child executions.
    Count { include_derived: bool },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeploymentStatus {
    Inactive,
    /// Queued to become Active on the next server restart.
    Enqueued,
    Active,
}

/// Outcome of [`DbExternalApi::enqueue_deployment`], reflecting what the transaction did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EnqueueOutcome {
    /// The target was inactive and is now Enqueued for the next restart.
    Enqueued,
    /// The target was already Active; it stays Active and any previously Enqueued
    /// deployment was cleared.
    AlreadyActive,
}

impl DeploymentStatus {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            DeploymentStatus::Inactive => "inactive",
            DeploymentStatus::Enqueued => "enqueued",
            DeploymentStatus::Active => "active",
        }
    }
}

impl std::str::FromStr for DeploymentStatus {
    type Err = StrVariant;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "inactive" => Ok(DeploymentStatus::Inactive),
            "enqueued" => Ok(DeploymentStatus::Enqueued),
            "active" => Ok(DeploymentStatus::Active),
            _ => Err(StrVariant::from(format!("unknown deployment status: {s}"))),
        }
    }
}

#[derive(Debug, Clone)]
pub struct DeploymentRecord {
    pub deployment_id: DeploymentId,
    pub description: Option<String>,
    /// Content digest = `sha256(deployment_toml)`.
    pub digest: ContentDigest,
    pub created_at: DateTime<Utc>,
    /// Set when the deployment becomes Active; None if it has never been active.
    pub last_active_at: Option<DateTime<Utc>>,
    pub status: DeploymentStatus,
    pub deployment_toml: String, // `deployment.toml` manifest that client enriched with generated metadata like `content_digest`, see `prepare_deployment_manifest`.
    pub obelisk_version: String,
    pub created_by: Option<String>,
    pub files: Vec<DeploymentFileRecord>,
}

impl DeploymentRecord {
    /// Computes the deployment content digest = `sha256(deployment_toml)`.
    #[must_use]
    pub fn compute_digest(deployment_toml: &str) -> ContentDigest {
        use sha2::{Digest as _, Sha256};
        let hash: [u8; 32] = Sha256::digest(deployment_toml.as_bytes()).into();
        ContentDigest(crate::component_id::Digest(hash))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeploymentFileRecord {
    pub path: String,
    pub digest: ContentDigest,
    pub size: u64,
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    strum::Display,
    strum::EnumString,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum ComponentFileRole {
    WasmComponent,
    ExecProgram,
    JsEntrypoint,
    JsModule,
    BacktraceSource,
    WitSource,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeploymentComponentFileRecord {
    pub component_name: StrVariant,
    pub path: String,
    pub role: ComponentFileRole,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeploymentComponentFileDetail {
    pub file: DeploymentFileRecord,
    pub role: ComponentFileRole,
}

/// Origin of a component's WIT: parsed from WASM or synthesized from type wrappers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, derive_more::Display, derive_more::TryFrom)]
#[try_from(repr)]
#[repr(i16)]
pub enum WitOrigin {
    #[display("wasm")]
    Wasm = 1,
    #[display("synthesized")]
    Synthesized = 2,
    #[display("authored")]
    Authored = 3,
}

#[derive(Debug, Clone)]
pub struct ComponentMetadataRecord {
    pub component_digest: ComponentDigest,
    pub imports: Vec<PersistedFunctionMetadata>,
    pub exports: Vec<PersistedFunctionMetadata>,
    pub wit: String,
    pub wit_origin: WitOrigin,
}

/// Relation between a deployment and its components
#[derive(Debug, Clone)]
pub struct DeploymentComponentRecord {
    pub deployment_id: DeploymentId,
    pub component_name: StrVariant,
    pub component_digest: ComponentDigest,
    pub component_type: ComponentType,
}

#[derive(Debug, Clone)]
pub struct DeploymentComponentDetail {
    pub component_id: ComponentId,
    pub imports: Vec<PersistedFunctionMetadata>,
    pub exports: Vec<PersistedFunctionMetadata>,
    pub wit: String,
    pub files: Vec<DeploymentComponentFileDetail>,
}

#[derive(
    Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, schemars::JsonSchema,
)]
pub struct PersistedFunctionMetadata {
    pub ffqn: FunctionFqn,
    pub parameter_types: Vec<PersistedParameterType>,
    pub return_type: String,
    pub extension: Option<FunctionExtension>,
    pub submittable: bool,
}

#[derive(
    Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, schemars::JsonSchema,
)]
pub struct PersistedParameterType {
    pub name: String,
    pub wit_type: String,
}

impl From<FunctionMetadata> for PersistedFunctionMetadata {
    fn from(value: FunctionMetadata) -> Self {
        PersistedFunctionMetadata {
            ffqn: value.ffqn,
            parameter_types: value
                .parameter_types
                .0
                .into_iter()
                .map(|param| PersistedParameterType {
                    name: param.name.to_string(),
                    wit_type: param.wit_type.to_string(),
                })
                .collect(),
            return_type: value.return_type.wit_type().to_string(),
            extension: value.extension,
            submittable: value.submittable,
        }
    }
}

#[derive(Debug)]
pub struct ListLogsResponse {
    pub items: Vec<LogEntryRow>,
    pub next_page: Pagination<LogCursor>, // Newer logs can always arrive e.g. via replay
    pub prev_page: Option<Pagination<LogCursor>>, // None if we are already at the beginning
}

#[derive(Debug)]
pub struct LogFilter {
    show_logs: bool,
    show_streams: bool,
    levels: Vec<LogLevel>, // Only applied if `show_logs` = true, empty means return all levels.
    stream_types: Vec<LogStreamType>, // Only applied if `show_streams` = true, empty means return all stream types.
    created_after: Option<DateTime<Utc>>,
    created_before: Option<DateTime<Utc>>,
}
impl LogFilter {
    // Constructor for logs only
    #[must_use]
    pub fn show_logs(levels: Vec<LogLevel>) -> LogFilter {
        LogFilter {
            show_logs: true,
            show_streams: false,
            levels,
            stream_types: Vec::new(),
            created_after: None,
            created_before: None,
        }
    }
    // Constructor for streams only
    #[must_use]
    pub fn show_streams(stream_types: Vec<LogStreamType>) -> LogFilter {
        LogFilter {
            show_logs: false,
            show_streams: true,
            levels: Vec::new(),
            stream_types,
            created_after: None,
            created_before: None,
        }
    }
    // Constructor for both logs and streams
    #[must_use]
    pub fn show_combined(levels: Vec<LogLevel>, stream_types: Vec<LogStreamType>) -> LogFilter {
        LogFilter {
            show_logs: true,
            show_streams: true,
            levels,
            stream_types,
            created_after: None,
            created_before: None,
        }
    }
    // Getters
    #[must_use]
    pub fn should_show_logs(&self) -> bool {
        self.show_logs
    }
    #[must_use]
    pub fn should_show_streams(&self) -> bool {
        self.show_streams
    }
    #[must_use]
    pub fn levels(&self) -> &Vec<LogLevel> {
        &self.levels
    }
    #[must_use]
    pub fn stream_types(&self) -> &Vec<LogStreamType> {
        &self.stream_types
    }
    #[must_use]
    pub fn with_created_bounds(
        mut self,
        created_after: Option<DateTime<Utc>>,
        created_before: Option<DateTime<Utc>>,
    ) -> Self {
        self.created_after = created_after;
        self.created_before = created_before;
        self
    }
    #[must_use]
    pub fn created_after(&self) -> Option<DateTime<Utc>> {
        self.created_after
    }
    #[must_use]
    pub fn created_before(&self) -> Option<DateTime<Utc>> {
        self.created_before
    }
}

#[derive(Debug, Clone)]
pub struct ExecutionWithStateRequestsResponses {
    pub execution_with_state: ExecutionWithState,
    pub events: Vec<ExecutionEvent>,
    pub responses: Vec<ResponseWithCursor>,
    pub max_version: Version,
    pub max_cursor: ResponseCursor,
}

#[async_trait]
pub trait DbConnection: DbExecutor {
    /// Get execution log.
    async fn get(&self, execution_id: &ExecutionId) -> Result<ExecutionLog, DbErrorRead>;

    /// Execution ids whose `lifecycle` is `cancelling`, for the cancellation driver
    /// to advance. Ordered oldest-first, capped at `batch_size`. Unlike the executor
    /// pick-up queries this is not lock/pause guarded: cancellation proceeds
    /// regardless (cancel supersedes pause).
    async fn get_cancelling(&self, batch_size: u32) -> Result<Vec<ExecutionId>, DbErrorRead>;

    async fn append_delay_response(
        &self,
        created_at: DateTime<Utc>,
        execution_id: ExecutionId,
        join_set_id: JoinSetId,
        delay_id: DelayId,
        outcome: Result<(), ()>, // Successfully finished - `Ok(())` or cancelled - `Err(())`
    ) -> Result<AppendDelayResponseOutcome, DbErrorWrite>;

    /// Append a batch of events to an existing execution log.
    /// The batch must not contain [`ExecutionRequest::Created`].
    async fn append_batch(
        &self,
        current_time: DateTime<Utc>, // not persisted, can be used for unblocking `subscribe_to_pending`
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
    ) -> Result<AppendBatchResponse, DbErrorWrite>;

    /// Append a blocking one-off delay batch (`JoinSetCreate`, `DelayRequest`, `JoinNext`)
    /// for a delay that is already due (e.g. `sleep(now)`) together with its `DelayFinished`
    /// response, in a single transaction. This unblocks the `JoinNext` immediately, so the
    /// workflow resumes without a round trip through the expired-timers watcher.
    async fn append_batch_with_delay_response(
        &self,
        current_time: DateTime<Utc>, // not persisted, can be used for unblocking `subscribe_to_pending`
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
        join_set_id: JoinSetId,
        delay_id: DelayId,
    ) -> Result<AppendBatchResponse, DbErrorWrite>;

    /// Append one or more events to the parent execution log, and create zero or more child execution logs.
    /// The batch must not contain [`ExecutionRequest::Created`].
    async fn append_batch_create_new_execution(
        &self,
        current_time: DateTime<Utc>, // not persisted, can be used for unblocking `subscribe_to_pending`
        batch: Vec<AppendRequest>,   // must not contain [`ExecutionRequest::Created`] events
        execution_id: ExecutionId,
        version: Version,
        child_req: Vec<CreateRequest>,
        backtraces: Vec<BacktraceInfo>,
    ) -> Result<AppendBatchResponse, DbErrorWrite>;

    /// Get a single event specified by version. Impls may set `ExecutionEvent::backtrace_id` to `None`.
    async fn get_execution_event(
        &self,
        execution_id: &ExecutionId,
        version: &Version,
    ) -> Result<ExecutionEvent, DbErrorRead>;

    /// Idempotent stub response write. Appends a Finished event to the child execution
    /// and a response to the parent. If the child is already finished with the same retval,
    /// succeeds silently. If finished with a different retval, returns [`DbErrorStubResponse::StubConflict`].
    async fn upsert_stub_response(
        &self,
        execution_id: ExecutionIdDerived,
        version: Version,
        req: AppendRequest,
        response: AppendResponseToExecution,
        current_time: DateTime<Utc>,
    ) -> Result<(), DbErrorStubResponse>;

    #[instrument(skip(self))]
    async fn get_create_request(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<CreateRequest, DbErrorRead> {
        let execution_event = self
            .get_execution_event(execution_id, &Version::new(0))
            .await?;
        if let ExecutionRequest::Created {
            ffqn,
            params,
            parent,
            scheduled_at,
            component_id,
            deployment_id,
            metadata,
            scheduled_by,
        } = execution_event.event
        {
            Ok(CreateRequest {
                created_at: execution_event.created_at,
                execution_id: execution_id.clone(),
                ffqn,
                params,
                parent,
                scheduled_at,
                component_id,
                deployment_id,
                metadata,
                scheduled_by,
                paused: false,
            })
        } else {
            Err(DbErrorRead::Generic(DbErrorGeneric::Uncategorized {
                reason: "execution log must start with creation".into(),
                context: SpanTrace::capture(),
                source: None,
                loc: Location::caller(),
            }))
        }
    }

    async fn get_pending_state(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<ExecutionWithState, DbErrorRead>;

    /// Get currently expired locks and async timers (delay requests)
    async fn get_expired_timers(
        &self,
        at: DateTime<Utc>,
    ) -> Result<Vec<ExpiredTimer>, DbErrorGeneric>;

    /// Create a new execution log
    async fn create(&self, req: CreateRequest) -> Result<AppendResponse, DbErrorWrite>;

    /// Notification mechainism with no strict guarantees for getting notified when a new response arrives.
    /// Parameter `start_idx` must be at most be equal to current size of responses in the execution log.
    /// If no response arrives immediately and `subscription_end_fut` resolves,
    /// `SubscribeToResponsesError::SubscriptionEnded` is returned.
    /// Implementations with no pubsub support should use polling.
    /// Callers are expected to call this function in a loop with a reasonable timeout
    /// to support less stellar implementations.
    async fn subscribe_to_next_responses(
        &self,
        execution_id: &ExecutionId,
        last_response: ResponseCursor,
        subscription_end_fut: Pin<Box<dyn Future<Output = ResponseSubscriptionEnd> + Send>>,
    ) -> Result<Vec<ResponseWithCursor>, SubscribeToResponsesError>;

    /// First, attempt to fetch the finished value. If the execution is not finished yet, poll
    /// periodically or subscribe to db changes, racing with `timeout_fut`.
    /// Notification mechainism with no strict guarantees for getting the finished result.
    /// Implementations with no pubsub support should use polling.
    /// Callers are expected to call this function in a loop with a reasonable timeout
    /// to support less stellar implementations.
    async fn wait_for_finished_result(
        &self,
        execution_id: &ExecutionId,
        timeout_fut: Option<Pin<Box<dyn Future<Output = TimeoutOutcome> + Send>>>,
    ) -> Result<SupportedFunctionReturnValue, DbErrorReadWithTimeout>;

    async fn append_backtrace(&self, append: BacktraceInfo) -> Result<(), DbErrorWrite>;

    async fn append_backtrace_batch(
        &self,
        batch: Vec<BacktraceInfo>,
    ) -> Result<usize, DbErrorWrite>;

    async fn append_log(&self, row: LogInfoAppendRow) -> Result<(), DbErrorWrite>;

    async fn append_log_batch(&self, batch: &[LogInfoAppendRow]) -> Result<(), DbErrorWrite>;

    /// Returns `TimeoutOutcome::Timeout` if not in Finished state.
    #[cfg(feature = "test")]
    async fn get_finished_result(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<SupportedFunctionReturnValue, DbErrorReadWithTimeout> {
        self.wait_for_finished_result(
            execution_id,
            Some(Box::pin(std::future::ready(TimeoutOutcome::Timeout))),
        )
        .await
    }
}

#[derive(Clone, Debug)]
pub struct LogInfoAppendRow {
    pub execution_id: ExecutionId,
    pub run_id: RunId,
    pub log_entry: LogEntry,
}

#[derive(Debug, Clone)]
pub struct LogEntryRow {
    pub cursor: LogCursor,
    pub run_id: RunId,
    pub log_entry: LogEntry,
    pub execution_id: ExecutionId,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct LogCursor(pub i64);

#[derive(Debug, Clone)]
pub enum LogEntry {
    Log {
        created_at: DateTime<Utc>,
        level: LogLevel,
        message: String,
    },
    Stream {
        created_at: DateTime<Utc>,
        payload: Vec<u8>,
        stream_type: LogStreamType,
    },
}
impl LogEntry {
    #[must_use]
    pub fn created_at(&self) -> DateTime<Utc> {
        match self {
            LogEntry::Log { created_at, .. } | LogEntry::Stream { created_at, .. } => *created_at,
        }
    }
}

#[derive(
    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, derive_more::TryFrom, strum::EnumIter,
)]
#[try_from(repr)]
#[repr(u8)]
pub enum LogLevel {
    Trace = 1,
    Debug,
    Info,
    Warn,
    Error,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, derive_more::TryFrom, strum::EnumIter)]
#[try_from(repr)]
#[repr(u8)]
pub enum LogStreamType {
    StdOut = 1,
    StdErr,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TimeoutOutcome {
    Timeout,
    Cancel,
}

#[cfg(feature = "test")]
#[async_trait]
pub trait DbConnectionTest: DbConnection {
    async fn append_response(
        &self,
        created_at: DateTime<Utc>,
        execution_id: ExecutionId,
        response_event: JoinSetResponseEvent,
    ) -> Result<(), DbErrorWrite>;
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CancelOutcome {
    CancelRequested,
    AlreadyFinished,
    AlreadyCancelling,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DelayCancelOutcome {
    Cancelled,
    AlreadyFinished,
}

#[instrument(skip(db_connection))]
pub async fn stub_execution(
    db_connection: &dyn DbConnection,
    execution_id: ExecutionIdDerived,
    parent_execution_id: ExecutionId,
    join_set_id: JoinSetId,
    created_at: DateTime<Utc>,
    return_value: SupportedFunctionReturnValue,
) -> Result<(), DbErrorWrite> {
    let stub_finished_version = Version::new(1); // Stub activities have no execution log except Created event.
    let finished_req = AppendRequest {
        created_at,
        event: ExecutionRequest::Finished {
            retval: return_value.clone(),
            http_client_traces: None,
        },
    };
    db_connection
        .upsert_stub_response(
            execution_id.clone(),
            stub_finished_version.clone(),
            finished_req,
            AppendResponseToExecution {
                parent_execution_id,
                created_at,
                join_set_id,
                child_execution_id: execution_id,
                finished_version: stub_finished_version,
                result: return_value,
            },
            created_at,
        )
        .await
        .map_err(|err| match err {
            DbErrorStubResponse::StubConflict => {
                DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::Conflict)
            }
            DbErrorStubResponse::Write(db_err) => db_err,
        })
}

pub async fn cancel_delay(
    db_connection: &dyn DbConnection,
    delay_id: DelayId,
    cancelled_at: DateTime<Utc>,
) -> Result<DelayCancelOutcome, DbErrorWrite> {
    let (parent_execution_id, join_set_id) = delay_id.split_to_parts();
    db_connection
        .append_delay_response(
            cancelled_at,
            parent_execution_id,
            join_set_id,
            delay_id,
            Err(()), // Mark as cancelled.
        )
        .await
        .map(|ok| match ok {
            AppendDelayResponseOutcome::Success | AppendDelayResponseOutcome::AlreadyCancelled => {
                DelayCancelOutcome::Cancelled
            }
            AppendDelayResponseOutcome::AlreadyFinished => DelayCancelOutcome::AlreadyFinished,
        })
}

#[derive(Clone, Debug)]
pub enum BacktraceFilter {
    First,
    Last,
    Specific(Version),
}

#[derive(Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "test", derive(Serialize))]
pub struct BacktraceInfo {
    pub execution_id: ExecutionId,
    pub component_id: ComponentId,
    pub version_min_including: Version,
    pub version_max_excluding: Version,
    pub wasm_backtrace: WasmBacktrace,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, schemars::JsonSchema)]
pub struct WasmBacktrace {
    pub frames: Vec<FrameInfo>,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, schemars::JsonSchema)]
pub struct FrameInfo {
    pub module: String,
    pub func_name: String,
    pub symbols: Vec<FrameSymbol>,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, schemars::JsonSchema)]
pub struct FrameSymbol {
    pub func_name: Option<String>,
    pub file: Option<String>,
    pub line: Option<u32>,
    pub col: Option<u32>,
}

mod wasm_backtrace {
    use super::{FrameInfo, FrameSymbol, WasmBacktrace};

    impl WasmBacktrace {
        pub fn maybe_from(backtrace: &wasmtime::WasmBacktrace) -> Option<Self> {
            if backtrace.frames().is_empty() {
                None
            } else {
                Some(Self {
                    frames: backtrace.frames().iter().map(FrameInfo::from).collect(),
                })
            }
        }
    }

    impl From<&wasmtime::FrameInfo> for FrameInfo {
        fn from(frame: &wasmtime::FrameInfo) -> Self {
            let module_name = frame.module().name().unwrap_or("<unknown>").to_string();
            let mut func_name = String::new();
            wasmtime_environ::demangle_function_name_or_index(
                &mut func_name,
                frame.func_name(),
                frame.func_index() as usize,
            )
            .expect("writing to string must succeed");
            Self {
                module: module_name,
                func_name,
                symbols: frame
                    .symbols()
                    .iter()
                    .map(std::convert::Into::into)
                    .collect(),
            }
        }
    }

    impl From<&wasmtime::FrameSymbol> for FrameSymbol {
        fn from(symbol: &wasmtime::FrameSymbol) -> Self {
            let func_name = symbol.name().map(|name| {
                let mut writer = String::new();
                wasmtime_environ::demangle_function_name(&mut writer, name)
                    .expect("writing to string must succeed");
                writer
            });

            Self {
                func_name,
                file: symbol.file().map(ToString::to_string),
                line: symbol.line(),
                col: symbol.column(),
            }
        }
    }
}
#[derive(Debug, Clone, derive_more::Display)]
#[display("{execution_id} {pending_state} {component_digest}")]
pub struct ExecutionWithState {
    pub execution_id: ExecutionId,
    pub ffqn: FunctionFqn,
    pub pending_state: PendingState,
    pub created_at: DateTime<Utc>,
    pub first_scheduled_at: DateTime<Utc>,
    pub component_digest: ComponentDigest,
    pub component_type: ComponentType,
    pub deployment_id: DeploymentId,
}

#[derive(Debug, Clone)]
pub enum ExecutionListPagination {
    CreatedBy(Pagination<Option<DateTime<Utc>>>),
    ExecutionId(Pagination<Option<ExecutionId>>),
}
impl Default for ExecutionListPagination {
    fn default() -> ExecutionListPagination {
        ExecutionListPagination::CreatedBy(Pagination::OlderThan {
            length: NonZeroU16::new(20).unwrap(),
            cursor: None,
            including_cursor: false, // does not matter when `cursor` is not specified
        })
    }
}
impl ExecutionListPagination {
    #[must_use]
    pub fn length(&self) -> u16 {
        match self {
            ExecutionListPagination::CreatedBy(pagination) => pagination.length(),
            ExecutionListPagination::ExecutionId(pagination) => pagination.length(),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Pagination<T> {
    NewerThan {
        length: NonZeroU16,
        cursor: T,
        including_cursor: bool,
    },
    OlderThan {
        length: NonZeroU16,
        cursor: T,
        including_cursor: bool,
    },
}
impl<T: Clone> Pagination<T> {
    pub fn length(&self) -> u16 {
        match self {
            Pagination::NewerThan { length, .. } | Pagination::OlderThan { length, .. } => {
                length.get()
            }
        }
    }

    pub fn length_nonzero(&self) -> NonZeroU16 {
        match self {
            Pagination::NewerThan { length, .. } | Pagination::OlderThan { length, .. } => *length,
        }
    }

    pub fn rel(&self) -> &'static str {
        match self {
            Pagination::NewerThan {
                including_cursor: false,
                ..
            } => ">",
            Pagination::NewerThan {
                including_cursor: true,
                ..
            } => ">=",
            Pagination::OlderThan {
                including_cursor: false,
                ..
            } => "<",
            Pagination::OlderThan {
                including_cursor: true,
                ..
            } => "<=",
        }
    }

    pub fn is_desc(&self) -> bool {
        matches!(self, Pagination::OlderThan { .. })
    }

    pub fn asc_or_desc(&self) -> &'static str {
        if self.is_asc() { "asc" } else { "desc" }
    }

    pub fn is_asc(&self) -> bool {
        !self.is_desc()
    }

    pub fn cursor(&self) -> &T {
        match self {
            Pagination::NewerThan { cursor, .. } | Pagination::OlderThan { cursor, .. } => cursor,
        }
    }

    #[must_use]
    pub fn invert(&self) -> Self {
        match self {
            Pagination::NewerThan {
                length,
                cursor,
                including_cursor,
            } => Pagination::OlderThan {
                length: *length,
                cursor: cursor.clone(),
                including_cursor: !including_cursor,
            },
            Pagination::OlderThan {
                length,
                cursor,
                including_cursor,
            } => Pagination::NewerThan {
                length: *length,
                cursor: cursor.clone(),
                including_cursor: !including_cursor,
            },
        }
    }
}

#[cfg(feature = "test")]
pub async fn wait_for_pending_state_fn<T: Debug>(
    db_connection: &dyn DbConnectionTest,
    execution_id: &ExecutionId,
    predicate: impl Fn(ExecutionLog) -> Option<T> + Send,
    timeout: Option<Duration>,
) -> Result<T, DbErrorReadWithTimeout> {
    tracing::trace!(%execution_id, "Waiting for predicate");
    let fut = async move {
        loop {
            let execution_log = db_connection.get(execution_id).await?;
            if let Some(t) = predicate(execution_log) {
                tracing::debug!(%execution_id, "Found: {t:?}");
                return Ok(t);
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    };

    if let Some(timeout) = timeout {
        tokio::select! { // future's liveness: Dropping the loser immediately.
            res = fut => res,
            () = tokio::time::sleep(timeout) => Err(DbErrorReadWithTimeout::Timeout(TimeoutOutcome::Timeout))
        }
    } else {
        fut.await
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ExpiredTimer {
    Lock(ExpiredLock),
    Delay(ExpiredDelay),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExpiredLock {
    pub execution_id: ExecutionId,
    // Version of last `Locked` event, used to detect whether the execution made progress.
    pub locked_at_version: Version,
    pub next_version: Version,
    /// As the execution may still be running, this represents the number of intermittent failures + timeouts prior to this execution.
    pub intermittent_event_count: u32,
    pub max_retries: Option<u32>,
    pub retry_exp_backoff: Duration,
    pub locked_by: LockedBy,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExpiredDelay {
    pub execution_id: ExecutionId,
    pub join_set_id: JoinSetId,
    pub delay_id: DelayId,
}

#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum PendingState {
    /// Caused by [`ExecutionRequest::Locked`].
    Locked(PendingStateLocked),

    #[display("PendingAt(`{_0}`)")]
    PendingAt(PendingStatePendingAt),

    /// Caused by [`HistoryEvent::JoinNext`]
    #[display("BlockedByJoinSet({_0})")]
    BlockedByJoinSet(PendingStateBlockedByJoinSet),

    /// Started by appending [`ExecutionRequest::Paused`] and
    /// ended with [`ExecutionRequest::Unpaused`] or [`ExecutionRequest::CancellationRequested`].
    ///
    /// Activity must not be in-flight, cancelling or finished when pausing.
    /// Workflow must not be cancelling or finished.
    /// Pausing a locked workflow must first append `Unlocked`.
    /// The previous pending state is stored for workflow unpause.
    #[display("Paused({_0})")]
    Paused(PendingStatePaused),

    /// Started by appending [`ExecutionRequest::CancellationRequested`] and
    /// ended with [`ExecutionRequest::Finished`].
    #[display("Cancelling({_0})")]
    Cancelling(PendingStateCancelling),

    #[display("Finished: {_0}")]
    Finished(PendingStateFinished),
}

/// [`PendingState`] flattened so the underlying runnable state and the
/// [`Lifecycle`] overlay are available side by side.
pub enum PendingStateMerged {
    Locked {
        state: PendingStateLocked,
        lifecycle: Lifecycle,
    },
    PendingAt {
        state: PendingStatePendingAt,
        lifecycle: Lifecycle,
    },
    BlockedByJoinSet {
        state: PendingStateBlockedByJoinSet,
        lifecycle: Lifecycle,
    },
    Finished(PendingStateFinished),
}
impl From<PendingState> for PendingStateMerged {
    fn from(state: PendingState) -> Self {
        match state {
            PendingState::Locked(s) => PendingStateMerged::Locked {
                state: s,
                lifecycle: Lifecycle::Active,
            },

            PendingState::PendingAt(s) => PendingStateMerged::PendingAt {
                state: s,
                lifecycle: Lifecycle::Active,
            },

            PendingState::BlockedByJoinSet(s) => PendingStateMerged::BlockedByJoinSet {
                state: s,
                lifecycle: Lifecycle::Active,
            },

            PendingState::Paused(inner) => match inner {
                PendingStatePaused::PendingAt(s) => PendingStateMerged::PendingAt {
                    state: s,
                    lifecycle: Lifecycle::Paused,
                },
                PendingStatePaused::BlockedByJoinSet(s) => PendingStateMerged::BlockedByJoinSet {
                    state: s,
                    lifecycle: Lifecycle::Paused,
                },
            },

            PendingState::Cancelling(inner) => match inner {
                PendingStateCancelling::Locked(s) => PendingStateMerged::Locked {
                    state: s,
                    lifecycle: Lifecycle::Cancelling,
                },
                PendingStateCancelling::PendingAt(s) => PendingStateMerged::PendingAt {
                    state: s,
                    lifecycle: Lifecycle::Cancelling,
                },
                PendingStateCancelling::BlockedByJoinSet(s) => {
                    PendingStateMerged::BlockedByJoinSet {
                        state: s,
                        lifecycle: Lifecycle::Cancelling,
                    }
                }
            },

            PendingState::Finished(s) => PendingStateMerged::Finished(s),
        }
    }
}

#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[display("Locked(`{lock_expires_at}`, {}, {})", locked_by.executor_id, locked_by.run_id)]
pub struct PendingStateLocked {
    pub locked_by: LockedBy,
    pub lock_expires_at: DateTime<Utc>,
}

#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[display("`{scheduled_at}`, last_lock={last_lock:?}")]
pub struct PendingStatePendingAt {
    pub scheduled_at: DateTime<Utc>,
    /// `last_lock` is needed for lock extension.
    pub last_lock: Option<LockedBy>,
}

#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[display("{join_set_id}, `{lock_expires_at}`, closing={closing}")]
pub struct PendingStateBlockedByJoinSet {
    pub join_set_id: JoinSetId,
    /// See [`HistoryEvent::JoinNext::lock_expires_at`].
    pub lock_expires_at: DateTime<Utc>,
    /// Blocked by closing of the join set
    pub closing: bool,
}

/// State of execution before it was paused.
///
/// A paused activity is always `PendingAt`. Pausing a locked workflow must first
/// append `Unlocked`, so `Locked` is never wrapped here.
#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
pub enum PendingStatePaused {
    #[display("PendingAt({_0})")]
    PendingAt(PendingStatePendingAt),
    #[display("BlockedByJoinSet({_0})")]
    BlockedByJoinSet(PendingStateBlockedByJoinSet),
}

/// Underlying state of a cancelling execution.
///
/// Tracked for the cancellation driver: an activity whose worker fails to confirm
/// teardown is pronounced finished once the `Locked` lease expires. The other
/// variants are a frozen snapshot from when cancellation was requested (incoming
/// responses do not unblock a cancelling execution), kept for observability.
#[derive(
    Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
pub enum PendingStateCancelling {
    #[display("Locked({_0})")]
    Locked(PendingStateLocked),
    #[display("PendingAt({_0})")]
    PendingAt(PendingStatePendingAt),
    #[display("BlockedByJoinSet({_0})")]
    BlockedByJoinSet(PendingStateBlockedByJoinSet),
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct LockedBy {
    pub executor_id: ExecutorId,
    pub run_id: RunId,
}
impl From<&Locked> for LockedBy {
    fn from(value: &Locked) -> Self {
        LockedBy {
            executor_id: value.executor_id,
            run_id: value.run_id,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct PendingStateFinished {
    pub version: VersionType, // not Version since it must be Copy
    pub finished_at: DateTime<Utc>,
    pub result_kind: PendingStateFinishedResultKind,
}
impl Display for PendingStateFinished {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.result_kind {
            PendingStateFinishedResultKind::Ok => write!(f, "OK"),
            PendingStateFinishedResultKind::Err(err) => write!(f, "{err}"),
        }
    }
}

// This is not a Result so that it can be customized for serialization
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum PendingStateFinishedResultKind {
    Ok,
    Err(PendingStateFinishedError),
}
impl PendingStateFinishedResultKind {
    pub fn as_result(&self) -> Result<(), &PendingStateFinishedError> {
        match self {
            PendingStateFinishedResultKind::Ok => Ok(()),
            PendingStateFinishedResultKind::Err(err) => Err(err),
        }
    }
}

impl From<&SupportedFunctionReturnValue> for PendingStateFinishedResultKind {
    fn from(result: &SupportedFunctionReturnValue) -> Self {
        result.as_pending_state_finished_result()
    }
}

#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    derive_more::Display,
    schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum PendingStateFinishedError {
    #[display("Execution failure ({_0})")]
    ExecutionFailure(ExecutionFailureKind),
    #[display("Error")]
    Error,
}

impl PendingState {
    #[instrument(skip(self))]
    pub fn can_append_lock(
        &self,
        created_at: DateTime<Utc>,
        executor_id: ExecutorId,
        run_id: RunId,
        lock_expires_at: DateTime<Utc>,
    ) -> Result<LockKind, DbErrorWriteNonRetriable> {
        if lock_expires_at <= created_at {
            return Err(DbErrorWriteNonRetriable::ValidationFailed(
                "invalid expiry date".into(),
            ));
        }
        match self {
            PendingState::PendingAt(PendingStatePendingAt {
                scheduled_at,
                last_lock,
            }) => {
                if *scheduled_at <= created_at {
                    // pending now, ok to lock
                    Ok(LockKind::CreatingNewLock)
                } else if let Some(LockedBy {
                    executor_id: last_executor_id,
                    run_id: last_run_id,
                }) = last_lock
                    && executor_id == *last_executor_id
                    && run_id == *last_run_id
                {
                    // Original executor is extending the lock.
                    Ok(LockKind::Extending)
                } else {
                    Err(DbErrorWriteNonRetriable::ValidationFailed(
                        "cannot lock, not yet pending".into(),
                    ))
                }
            }
            PendingState::Locked(PendingStateLocked {
                locked_by:
                    LockedBy {
                        executor_id: current_pending_state_executor_id,
                        run_id: current_pending_state_run_id,
                    },
                lock_expires_at: _,
            }) => {
                if executor_id == *current_pending_state_executor_id
                    && run_id == *current_pending_state_run_id
                {
                    // Original executor is extending the lock.
                    Ok(LockKind::Extending)
                } else {
                    Err(DbErrorWriteNonRetriable::IllegalState {
                        reason: "cannot lock, already locked".into(),
                        context: SpanTrace::capture(),
                        source: None,
                        loc: Location::caller(),
                    })
                }
            }
            PendingState::BlockedByJoinSet { .. } => Err(DbErrorWriteNonRetriable::IllegalState {
                reason: "cannot append Locked event when in BlockedByJoinSet state".into(),
                context: SpanTrace::capture(),
                source: None,
                loc: Location::caller(),
            }),
            PendingState::Finished { .. } => Err(DbErrorWriteNonRetriable::IllegalState {
                reason: "already finished".into(),
                context: SpanTrace::capture(),
                source: None,
                loc: Location::caller(),
            }),
            PendingState::Paused(..) => Err(DbErrorWriteNonRetriable::IllegalState {
                reason: "cannot lock, execution is paused".into(),
                context: SpanTrace::capture(),
                source: None,
                loc: Location::caller(),
            }),
            PendingState::Cancelling(..) => Err(DbErrorWriteNonRetriable::IllegalState {
                reason: "cannot lock, execution is cancelling".into(),
                context: SpanTrace::capture(),
                source: None,
                loc: Location::caller(),
            }),
        }
    }

    #[must_use]
    pub fn is_finished(&self) -> bool {
        matches!(self, PendingState::Finished { .. })
    }

    #[must_use]
    pub fn is_paused(&self) -> bool {
        matches!(self, PendingState::Paused(_))
    }

    #[must_use]
    pub fn is_cancelling(&self) -> bool {
        matches!(self, PendingState::Cancelling(_))
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LockKind {
    Extending,
    CreatingNewLock,
}

pub mod http_client_trace {
    use chrono::{DateTime, Utc};
    use serde::{Deserialize, Serialize};

    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
    pub struct HttpClientTrace {
        pub req: RequestTrace,
        pub resp: Option<ResponseTrace>,
    }

    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
    pub struct RequestTrace {
        pub sent_at: DateTime<Utc>,
        pub uri: String,
        pub method: String,
    }

    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
    pub struct ResponseTrace {
        pub finished_at: DateTime<Utc>,
        pub status: Result<u16, String>,
    }
}

/// Root type for DB schema generation - contains all serialized DB types
#[derive(schemars::JsonSchema)]
pub struct DbStorageSchema {
    pub execution_event: ExecutionEvent,
    pub pending_state: PendingState,
    pub join_set_response: JoinSetResponse,
    pub wasm_backtrace: WasmBacktrace,
    pub persisted_function_metadata: PersistedFunctionMetadata,
}

#[cfg(test)]
mod tests {
    use super::HistoryEvent;
    use super::HistoryEventScheduleAt;
    use super::JoinNextTryOutcome;
    use super::PendingStateFinished;
    use super::PendingStateFinishedError;
    use super::PendingStateFinishedResultKind;
    use crate::ExecutionFailureKind;
    use crate::JoinSetId;
    use crate::SupportedFunctionReturnValue;
    use chrono::DateTime;
    use chrono::Datelike;
    use insta::assert_snapshot;
    use rstest::rstest;
    use std::time::Duration;
    use val_json::type_wrapper::TypeWrapper;
    use val_json::wast_val::WastVal;
    use val_json::wast_val::WastValWithType;

    #[rstest(expected => [
        PendingStateFinishedResultKind::Ok,
        PendingStateFinishedResultKind::Err(PendingStateFinishedError::ExecutionFailure(ExecutionFailureKind::TimedOut)),
    ])]
    #[test]
    fn serde_pending_state_finished_result_kind_should_work(
        expected: PendingStateFinishedResultKind,
    ) {
        let ser = serde_json::to_string(&expected).unwrap();
        let actual: PendingStateFinishedResultKind = serde_json::from_str(&ser).unwrap();
        assert_eq!(expected, actual);
    }

    #[test]
    fn result_kind_json_constants_match_serde() {
        assert_eq!(
            crate::storage::RESULT_KIND_JSON_OK,
            serde_json::to_string(&PendingStateFinishedResultKind::Ok).unwrap()
        );
        assert_eq!(
            crate::storage::RESULT_KIND_JSON_ERROR,
            serde_json::to_string(&PendingStateFinishedResultKind::Err(
                PendingStateFinishedError::Error
            ))
            .unwrap()
        );
    }

    #[rstest(result_kind => [
        PendingStateFinishedResultKind::Ok,
        PendingStateFinishedResultKind::Err(PendingStateFinishedError::ExecutionFailure(ExecutionFailureKind::TimedOut)),
    ])]
    #[test]
    fn serde_pending_state_finished_should_work(result_kind: PendingStateFinishedResultKind) {
        let expected = PendingStateFinished {
            version: 0,
            finished_at: DateTime::UNIX_EPOCH,
            result_kind,
        };

        let ser = serde_json::to_string(&expected).unwrap();
        let actual: PendingStateFinished = serde_json::from_str(&ser).unwrap();
        assert_eq!(expected, actual);
    }

    #[test]
    fn join_set_deser_with_result_ok_option_none_should_work() {
        let expected = SupportedFunctionReturnValue::Ok(Some(WastValWithType {
            r#type: TypeWrapper::Result {
                ok: Some(Box::new(TypeWrapper::Option(Box::new(TypeWrapper::String)))),
                err: Some(Box::new(TypeWrapper::String)),
            },
            value: WastVal::Result(Ok(Some(Box::new(WastVal::Option(None))))),
        }));
        let json = serde_json::to_string(&expected).unwrap();
        assert_snapshot!(json);

        let actual: SupportedFunctionReturnValue = serde_json::from_str(&json).unwrap();

        assert_eq!(expected, actual);
    }

    #[test]
    fn as_date_time_should_work_with_duration_u32_max_secs() {
        let duration = Duration::from_secs(u64::from(u32::MAX));
        let schedule_at = HistoryEventScheduleAt::In(duration);
        let resolved = schedule_at.as_date_time(DateTime::UNIX_EPOCH).unwrap();
        assert_eq!(2106, resolved.year());
    }

    const MILLIS_PER_SEC: i64 = 1000;
    const TIMEDELTA_MAX_SECS: i64 = i64::MAX / MILLIS_PER_SEC;

    #[test]
    fn as_date_time_should_fail_on_duration_secs_greater_than_i64_max() {
        // Fails on duration -> timedelta conversion, but a smaller duration can fail on datetime + timedelta
        let duration = Duration::from_secs(
            u64::try_from(TIMEDELTA_MAX_SECS).expect("positive number must not fail") + 1,
        );
        let schedule_at = HistoryEventScheduleAt::In(duration);
        schedule_at.as_date_time(DateTime::UNIX_EPOCH).unwrap_err();
    }

    #[test]
    fn join_next_try_outcome_new_format() {
        let json = r#"{"type":"join_next_try","join_set_id":"n:test","outcome":"found"}"#;
        let event: HistoryEvent = serde_json::from_str(json).unwrap();
        assert_eq!(
            event,
            HistoryEvent::JoinNextTry {
                join_set_id: JoinSetId::new(
                    crate::JoinSetKind::Named,
                    crate::StrVariant::Static("test")
                )
                .unwrap(),
                outcome: JoinNextTryOutcome::Found,
            }
        );

        let json = r#"{"type":"join_next_try","join_set_id":"n:test","outcome":"all_processed"}"#;
        let event: HistoryEvent = serde_json::from_str(json).unwrap();
        assert_eq!(
            event,
            HistoryEvent::JoinNextTry {
                join_set_id: JoinSetId::new(
                    crate::JoinSetKind::Named,
                    crate::StrVariant::Static("test")
                )
                .unwrap(),
                outcome: JoinNextTryOutcome::AllProcessed,
            }
        );
    }

    #[test]
    fn join_next_try_outcome_serializes_new_format() {
        let event = HistoryEvent::JoinNextTry {
            join_set_id: JoinSetId::new(
                crate::JoinSetKind::Named,
                crate::StrVariant::Static("test"),
            )
            .unwrap(),
            outcome: JoinNextTryOutcome::AllProcessed,
        };
        let json = serde_json::to_string(&event).unwrap();
        assert!(
            json.contains(r#""outcome":"all_processed""#),
            "expected outcome field, got: {json}"
        );
        assert!(
            !json.contains("found_response"),
            "should not contain old field, got: {json}"
        );
    }

    mod stub_retval_hash {
        use super::super::{StubRetVal, StubRetValHash};
        use crate::SupportedFunctionReturnValue;
        use val_json::type_wrapper::TypeWrapper;
        use val_json::wast_val::{WastVal, WastValWithType};

        #[test]
        fn typed_variant_hash_is_stable() {
            let retval =
                StubRetVal::Typed(SupportedFunctionReturnValue::Ok(Some(WastValWithType {
                    r#type: TypeWrapper::String,
                    value: WastVal::String("hello".into()),
                })));
            let hash = retval.hash();
            // Hash should start with version byte 0x01
            assert_eq!(hash.to_string().chars().take(2).collect::<String>(), "01");
            // Hash should be 66 hex characters (33 bytes * 2)
            assert_eq!(hash.to_string().len(), 66);
        }

        #[test]
        fn untyped_variant_hash_is_stable() {
            let retval = StubRetVal::Untyped(r#"{"ok": "hello"}"#.to_string());
            let hash = retval.hash();
            // Hash should start with version byte 0x01
            assert_eq!(hash.to_string().chars().take(2).collect::<String>(), "01");
            // Hash should be 66 hex characters (33 bytes * 2)
            assert_eq!(hash.to_string().len(), 66);
        }

        #[test]
        fn different_values_produce_different_hashes() {
            let typed1 = StubRetVal::Typed(SupportedFunctionReturnValue::Ok(None));
            let typed2 = StubRetVal::Typed(SupportedFunctionReturnValue::Err(None));
            let untyped1 = StubRetVal::Untyped("value1".to_string());
            let untyped2 = StubRetVal::Untyped("value2".to_string());

            let hashes: Vec<_> = [typed1, typed2, untyped1, untyped2]
                .into_iter()
                .map(|r| r.hash().to_string())
                .collect();

            // All hashes should be unique
            for (i, h1) in hashes.iter().enumerate() {
                for h2 in hashes.iter().skip(i + 1) {
                    assert_ne!(h1, h2, "hashes should be different");
                }
            }
        }

        #[test]
        fn same_values_produce_same_hashes() {
            let retval1 = StubRetVal::Typed(SupportedFunctionReturnValue::Ok(None));
            let retval2 = StubRetVal::Typed(SupportedFunctionReturnValue::Ok(None));
            assert_eq!(retval1.hash(), retval2.hash());

            let untyped1 = StubRetVal::Untyped("test".to_string());
            let untyped2 = StubRetVal::Untyped("test".to_string());
            assert_eq!(untyped1.hash(), untyped2.hash());
        }

        #[test]
        fn hash_serialization_roundtrip() {
            let retval = StubRetVal::Typed(SupportedFunctionReturnValue::Ok(None));
            let hash = retval.hash();

            let serialized = serde_json::to_string(&hash).unwrap();
            let deserialized: StubRetValHash = serde_json::from_str(&serialized).unwrap();

            assert_eq!(hash, deserialized);
        }

        #[test]
        fn hash_display_and_fromstr_roundtrip() {
            let retval = StubRetVal::Untyped("test value".to_string());
            let hash = retval.hash();

            let display = hash.to_string();
            let parsed: StubRetValHash = display.parse().unwrap();

            assert_eq!(hash, parsed);
        }

        #[test]
        fn typed_and_untyped_with_same_content_produce_different_hashes() {
            // Even if the JSON content is the same, Typed vs Untyped should hash differently
            let typed = StubRetVal::Typed(SupportedFunctionReturnValue::Ok(None));
            let json_of_typed =
                serde_json::to_string(&SupportedFunctionReturnValue::Ok(None)).unwrap();
            let untyped = StubRetVal::Untyped(json_of_typed);

            assert_ne!(typed.hash(), untyped.hash());
        }
    }
}