obeli_sk_db_sqlite/

sqlite_dao.rs

use crate::{histograms::Histograms, sqlite_dao::conversions::to_generic_error};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use concepts::{
    ComponentId, ComponentRetryConfig, ComponentType, ContentDigest, ExecutionFailureKind,
    ExecutionId, FunctionFqn, JoinSetId, StrVariant, SupportedFunctionReturnValue,
    cas::{Cas, CasError},
    component_id::{ComponentDigest, Digest},
    prefixed_ulid::{DelayId, DeploymentId, ExecutionIdDerived, ExecutorId, RunId},
    storage::{
        AppendBatchResponse, AppendDelayResponseOutcome, AppendEventsToExecution, AppendRequest,
        AppendResponse, AppendResponseToExecution, BacktraceFilter, BacktraceInfo, CancelOutcome,
        ComponentMetadataRecord, ComponentUpgradeOutcome, ComponentUpgradeReason, CreateRequest,
        DUMMY_CREATED, DUMMY_HISTORY_EVENT, DbConnection, DbErrorGeneric, DbErrorRead,
        DbErrorReadWithTimeout, DbErrorStubResponse, DbErrorWrite, DbErrorWriteNonRetriable,
        DbExecutor, DbExternalApi, DbPool, DbPoolCloseable, DeploymentComponentDetail,
        DeploymentComponentRecord, DeploymentExecutionCounts, DeploymentFileRecord,
        DeploymentRecord, DeploymentState, DeploymentStatus, EnqueueOutcome, ExecutionEvent,
        ExecutionListPagination, ExecutionRequest, ExecutionWithState,
        ExecutionWithStateRequestsResponses, ExpiredDelay, ExpiredLock, ExpiredTimer,
        HISTORY_EVENT_TYPE_JOIN_NEXT, HistoryEvent, JoinSetRequest, JoinSetResponse,
        JoinSetResponseEvent, JoinSetResponseEventOuter, LIFECYCLE_ACTIVE, LIFECYCLE_CANCELLING,
        LIFECYCLE_PAUSED, Lifecycle, ListExecutionEventsResponse, ListExecutionsFilter,
        ListLogsResponse, ListResponsesResponse, LockPendingResponse, Locked, LockedBy,
        LockedExecution, LogCursor, LogEntry, LogEntryRow, LogFilter, LogInfoAppendRow, LogLevel,
        LogStreamType, Pagination, PendingState, PendingStateBlockedByJoinSet,
        PendingStateFinishedError, PendingStateFinishedResultKind, PendingStateMerged,
        RESULT_KIND_JSON_ERROR, RESULT_KIND_JSON_OK, ResponseCursor, ResponseWithCursor,
        STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED, STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome,
        Unlocked, Version, VersionType,
    },
};
use conversions::{JsonWrapper, consistency_db_err, consistency_rusqlite, from_generic_error};
use db_common::{
    AppendNotifier, CombinedState, CombinedStateDTO, NotifierExecutionFinished, NotifierPendingAt,
    PendingFfqnSubscribersHolder, state_filter_to_sql, state_filters_now,
};
use hashbrown::HashMap;
use rusqlite::{
    CachedStatement, Connection, OpenFlags, OptionalExtension, Row, ToSql, Transaction,
    TransactionBehavior, named_params,
    types::{ToSqlOutput, Value},
};
use sha2::{Digest as _, Sha256};
use std::{
    cmp::max,
    collections::VecDeque,
    fmt::Debug,
    ops::DerefMut,
    panic::Location,
    path::Path,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};
use std::{fmt::Write as _, pin::Pin, str::FromStr as _};
use strum::IntoEnumIterator as _;
use tokio::sync::{mpsc, oneshot};
use tracing::{Level, Span, debug, error, info, instrument, trace, warn};
use tracing_error::SpanTrace;

#[derive(Debug, thiserror::Error)]
#[error("initialization error")]
pub struct InitializationError;

#[derive(Debug, Clone)]
struct DelayReq {
    join_set_id: JoinSetId,
    delay_id: DelayId,
    expires_at: DateTime<Utc>,
    paused: bool,
}
/*
mmap_size = 128MB - Set the global memory map so all processes can share some data
https://www.sqlite.org/pragma.html#pragma_mmap_size
https://www.sqlite.org/mmap.html

journal_size_limit = 64 MB - limit on the WAL file to prevent unlimited growth
https://www.sqlite.org/pragma.html#pragma_journal_size_limit

Inspired by https://github.com/rails/rails/pull/49349
*/

const PRAGMA: [[&str; 2]; 10] = [
    ["journal_mode", "wal"],
    ["synchronous", "FULL"],
    ["foreign_keys", "true"],
    ["busy_timeout", "1000"],
    ["cache_size", "10000"], // number of pages
    ["temp_store", "MEMORY"],
    ["page_size", "8192"], // 8 KB
    ["mmap_size", "134217728"],
    ["journal_size_limit", "67108864"],
    ["integrity_check", ""],
];

mod embedded {
    refinery::embed_migrations!("migrations");
}

#[derive(Debug, thiserror::Error, Clone)]
enum RusqliteError {
    #[error("not found")]
    NotFound,
    #[error("generic: {reason}")]
    Generic {
        reason: StrVariant,
        context: SpanTrace,
        source: Option<Arc<dyn std::error::Error + Send + Sync>>,
        loc: &'static Location<'static>,
    },
    #[error("close")]
    Close,
}

mod conversions {

    use super::RusqliteError;
    use concepts::{
        StrVariant,
        storage::{
            DbErrorGeneric, DbErrorRead, DbErrorReadWithTimeout, DbErrorStubResponse, DbErrorWrite,
        },
    };
    use rusqlite::{
        ToSql,
        types::{FromSql, FromSqlError},
    };
    use std::{fmt::Debug, panic::Location, sync::Arc};
    use tracing::error;
    use tracing_error::SpanTrace;

    impl From<rusqlite::Error> for RusqliteError {
        // The LTX returns this, capture inner location.
        #[track_caller]
        fn from(err: rusqlite::Error) -> Self {
            if matches!(err, rusqlite::Error::QueryReturnedNoRows) {
                RusqliteError::NotFound
            } else {
                RusqliteError::Generic {
                    reason: err.to_string().into(),
                    context: SpanTrace::capture(),
                    source: Some(Arc::new(err)),
                    loc: Location::caller(),
                }
            }
        }
    }

    // Manual conversion, as this function ignores the NotFound.
    #[track_caller]
    pub fn to_generic_error(err: RusqliteError) -> DbErrorGeneric {
        if let RusqliteError::Close = err {
            DbErrorGeneric::Close
        } else {
            DbErrorGeneric::Uncategorized {
                reason: err.to_string().into(),
                context: SpanTrace::capture(),
                source: Some(Arc::new(err)),
                loc: Location::caller(),
            }
        }
    }

    /// Convert a `DbErrorGeneric` to `rusqlite::Error` for use in row mapping closures.
    #[track_caller]
    pub fn from_generic_error(err: &DbErrorGeneric) -> rusqlite::Error {
        FromSqlError::other(OtherError {
            reason: err.to_string().into(),
            loc: Location::caller(),
        })
        .into()
    }

    impl From<RusqliteError> for DbErrorRead {
        fn from(err: RusqliteError) -> Self {
            if matches!(err, RusqliteError::NotFound) {
                Self::NotFound
            } else {
                to_generic_error(err).into()
            }
        }
    }
    impl From<RusqliteError> for DbErrorReadWithTimeout {
        fn from(err: RusqliteError) -> Self {
            Self::from(DbErrorRead::from(err))
        }
    }
    impl From<RusqliteError> for DbErrorWrite {
        fn from(err: RusqliteError) -> Self {
            if matches!(err, RusqliteError::NotFound) {
                Self::NotFound
            } else {
                to_generic_error(err).into()
            }
        }
    }

    impl From<RusqliteError> for DbErrorStubResponse {
        fn from(err: RusqliteError) -> Self {
            DbErrorStubResponse::Write(DbErrorWrite::from(err))
        }
    }

    pub(crate) struct JsonWrapper<T>(pub(crate) T);
    impl<T: serde::de::DeserializeOwned + 'static + Debug> FromSql for JsonWrapper<T> {
        fn column_result(
            value: rusqlite::types::ValueRef<'_>,
        ) -> rusqlite::types::FromSqlResult<Self> {
            let value = match value {
                rusqlite::types::ValueRef::Text(value) | rusqlite::types::ValueRef::Blob(value) => {
                    Ok(value)
                }
                other => {
                    error!(
                        backtrace = %std::backtrace::Backtrace::capture(),
                        "Unexpected type when conveting to JSON - expected Text or Blob, got type `{other:?}`",
                    );
                    Err(FromSqlError::InvalidType)
                }
            }?;
            let value = serde_json::from_slice::<T>(value).map_err(|err| {
                error!(
                    backtrace = %std::backtrace::Backtrace::capture(),
                    "Cannot convert JSON value `{value:?}` to type:`{type}` - {err:?}",
                    r#type = std::any::type_name::<T>()
                );
                FromSqlError::InvalidType
            })?;
            Ok(Self(value))
        }
    }
    impl<T: serde::ser::Serialize + Debug> ToSql for JsonWrapper<T> {
        fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
            let string = serde_json::to_string(&self.0).map_err(|err| {
                error!(
                    "Cannot serialize {value:?} of type `{type}` - {err:?}",
                    value = self.0,
                    r#type = std::any::type_name::<T>()
                );
                rusqlite::Error::ToSqlConversionFailure(Box::new(err))
            })?;
            Ok(rusqlite::types::ToSqlOutput::Owned(
                rusqlite::types::Value::Text(string),
            ))
        }
    }

    // Used as a wrapper for `FromSqlError::Other`
    #[derive(Debug, thiserror::Error)]
    #[error("{reason}")]
    pub(crate) struct OtherError {
        reason: StrVariant,
        loc: &'static Location<'static>,
    }

    #[track_caller]
    pub(crate) fn consistency_rusqlite(reason: impl Into<StrVariant>) -> rusqlite::Error {
        FromSqlError::other(OtherError {
            reason: reason.into(),
            loc: Location::caller(),
        })
        .into()
    }

    #[track_caller]
    pub(crate) fn consistency_db_err(reason: impl Into<StrVariant>) -> DbErrorGeneric {
        DbErrorGeneric::Uncategorized {
            reason: reason.into(),
            context: SpanTrace::capture(),
            source: None,
            loc: Location::caller(),
        }
    }
}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum TxType {
    MultipleWrites, // PhyTx must be rolled back, other LTX restarted
    Other,          // Read only or a single write LTX. Continue the PhyTx if LTX returns error.
}

#[derive(Debug, Clone, Copy)]
enum CancellationFfqnCheck {
    Required,
    Skipped,
}

#[derive(Clone)]
struct CommitError(RusqliteError);

#[derive(Debug)]
struct ShouldRollback;

#[derive(derive_more::Debug)]
struct LogicalTx {
    #[debug(skip)]
    #[expect(clippy::type_complexity)]
    func: Box<dyn FnMut(&mut Transaction) -> Result<(), ShouldRollback> + Send>,
    sent_at: Instant,
    func_name: &'static str,
    #[debug(skip)]
    // Result of Physical Tx commit / rollback
    phytx_flush_sender: oneshot::Sender<Result<(), CommitError>>,
    priority: LtxPriority,
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum LtxPriority {
    High,
    Low,
}

#[derive(derive_more::Debug)]
enum ThreadCommand {
    LogicalTx(LogicalTx),
    Shutdown,
}

#[derive(Clone)]
pub struct SqlitePool(SqlitePoolInner);

/// Notification channel where subscriber blocks waiting for a response. Writer thread
/// ensures serializability - subscriber ensures there are no missed responses. Senders
/// notify outside of the writer thread, making ordering not guaranteed. Subscriber must
/// therefore refetch from its own watermark.
type ResponseSubscribers =
    Arc<Mutex<HashMap<ExecutionId, (oneshot::Sender<()>, u64 /* unique_tag */)>>>;
type PendingSubscribers = Arc<Mutex<PendingFfqnSubscribersHolder>>;
type ExecutionFinishedSubscribers = Mutex<
    HashMap<
        ExecutionId,
        HashMap<u64 /* unique_tag */, oneshot::Sender<SupportedFunctionReturnValue>>,
    >,
>;

#[derive(Clone)]
struct SqlitePoolInner {
    shutdown_requested: Arc<AtomicBool>,
    shutdown_finished: Arc<AtomicBool>,
    command_tx: tokio::sync::mpsc::Sender<ThreadCommand>,
    response_subscribers: ResponseSubscribers,
    pending_subscribers: PendingSubscribers,
    execution_finished_subscribers: Arc<ExecutionFinishedSubscribers>,
    join_handle: Option<Arc<std::thread::JoinHandle<()>>>, // always Some, Optional for swapping in drop.
}

#[async_trait]
impl DbPoolCloseable for SqlitePool {
    async fn close(&self) {
        debug!("Sqlite is closing");
        self.0.shutdown_requested.store(true, Ordering::Release);
        // Unblock the thread's blocking_recv. If the capacity is reached, the next processed message will trigger shutdown.
        let _ = self.0.command_tx.try_send(ThreadCommand::Shutdown);
        while !self.0.shutdown_finished.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(1)).await;
        }
        debug!("Sqlite was closed");
    }
}

#[async_trait]
impl DbPool for SqlitePool {
    async fn db_exec_conn(&self) -> Result<Box<dyn DbExecutor>, DbErrorGeneric> {
        if self.0.shutdown_requested.load(Ordering::Acquire) {
            return Err(DbErrorGeneric::Close);
        }
        Ok(Box::new(self.clone()))
    }

    async fn connection(&self) -> Result<Box<dyn DbConnection>, DbErrorGeneric> {
        if self.0.shutdown_requested.load(Ordering::Acquire) {
            return Err(DbErrorGeneric::Close);
        }
        Ok(Box::new(self.clone()))
    }
    async fn external_api_conn(&self) -> Result<Box<dyn DbExternalApi>, DbErrorGeneric> {
        if self.0.shutdown_requested.load(Ordering::Acquire) {
            return Err(DbErrorGeneric::Close);
        }
        Ok(Box::new(self.clone()))
    }
    async fn cas_conn(&self) -> Result<Box<dyn Cas>, DbErrorGeneric> {
        if self.0.shutdown_requested.load(Ordering::Acquire) {
            return Err(DbErrorGeneric::Close);
        }
        Ok(Box::new(self.clone()))
    }
    #[cfg(feature = "test")]
    async fn connection_test(
        &self,
    ) -> Result<Box<dyn concepts::storage::DbConnectionTest>, DbErrorGeneric> {
        if self.0.shutdown_requested.load(Ordering::Acquire) {
            return Err(DbErrorGeneric::Close);
        }
        Ok(Box::new(self.clone()))
    }
}

impl Drop for SqlitePool {
    fn drop(&mut self) {
        let arc = self.0.join_handle.take().expect("join_handle was set");
        if let Ok(join_handle) = Arc::try_unwrap(arc) {
            // Last holder
            if !join_handle.is_finished() {
                if !self.0.shutdown_finished.load(Ordering::Acquire) {
                    // Best effort to shut down the sqlite thread.
                    let backtrace = std::backtrace::Backtrace::capture();
                    warn!("SqlitePool was not closed properly - {backtrace}");
                    self.0.shutdown_requested.store(true, Ordering::Release);
                    // Unblock the thread's blocking_recv. If the capacity is reached, the next processed message will trigger shutdown.
                    let _ = self.0.command_tx.try_send(ThreadCommand::Shutdown);
                    // Not joining the thread, drop might be called from async context.
                    // We are shutting down the server anyway.
                } else {
                    // The thread set `shutdown_finished` as its last operation.
                }
            }
        }
    }
}

#[derive(Debug, Clone)]
pub struct SqliteConfig {
    pub queue_capacity: usize,
    pub pragma_override: Option<HashMap<String, String>>,
    pub metrics_threshold: Option<Duration>,
}
impl Default for SqliteConfig {
    fn default() -> Self {
        Self {
            queue_capacity: 100,
            pragma_override: None,
            metrics_threshold: None,
        }
    }
}

struct ShutdownRequested;

fn deployment_record_from_row(row: &Row<'_>) -> rusqlite::Result<DeploymentRecord> {
    let deployment_id: DeploymentId = row.get("deployment_id")?;
    let status_str: String = row.get("status")?;
    let status = status_str.parse::<DeploymentStatus>().map_err(|_| {
        rusqlite::Error::InvalidColumnType(3, "status".to_string(), rusqlite::types::Type::Text)
    })?;
    Ok(DeploymentRecord {
        deployment_id,
        description: row.get("description")?,
        digest: row.get("digest")?,
        created_at: row.get("created_at")?,
        last_active_at: row.get("last_active_at")?,
        status,
        deployment_toml: row.get("deployment_toml")?,
        obelisk_version: row.get("obelisk_version")?,
        created_by: row.get("created_by")?,
        files: Vec::new(),
    })
}

fn deployment_component_detail_from_row(
    row: &Row<'_>,
) -> Result<DeploymentComponentDetail, DbErrorGeneric> {
    let component_name: String = row
        .get("component_name")
        .map_err(|err| consistency_db_err(format!("invalid component_name: {err}")))?;
    let component_type: String = row
        .get("component_type")
        .map_err(|err| consistency_db_err(format!("invalid component_type: {err}")))?;
    let component_type = component_type
        .parse::<ComponentType>()
        .map_err(|err| consistency_db_err(format!("invalid component_type: {err}")))?;
    let component_digest: ComponentDigest = row
        .get("component_digest")
        .map_err(|err| consistency_db_err(format!("invalid component_digest: {err}")))?;
    let component_id = ComponentId::new(
        component_type,
        StrVariant::from(component_name),
        component_digest,
    )
    .map_err(|err| consistency_db_err(err.to_string()))?;
    let imports: String = row
        .get("imports_json")
        .map_err(|err| consistency_db_err(format!("invalid imports_json: {err}")))?;
    let imports = serde_json::from_str(&imports)
        .map_err(|err| consistency_db_err(format!("invalid imports_json: {err}")))?;
    let exports: String = row
        .get("exports_json")
        .map_err(|err| consistency_db_err(format!("invalid exports_json: {err}")))?;
    let exports = serde_json::from_str(&exports)
        .map_err(|err| consistency_db_err(format!("invalid exports_json: {err}")))?;
    let wit: String = row
        .get("wit")
        .map_err(|err| consistency_db_err(format!("invalid wit: {err}")))?;
    Ok(DeploymentComponentDetail {
        component_id,
        imports,
        exports,
        wit,
    })
}

struct PendingAfterEventUpdate {
    scheduled_at: DateTime<Utc>,
    intermittent_failure: bool,
    component_input_digest: ComponentDigest,
}

impl SqlitePool {
    fn init_thread(
        path: &Path,
        mut pragma_override: HashMap<String, String>,
    ) -> Result<Connection, InitializationError> {
        fn pragma_update(
            conn: &Connection,
            name: &str,
            value: &str,
        ) -> Result<(), InitializationError> {
            if value.is_empty() {
                debug!("Querying PRAGMA {name}");
                conn.pragma_query(None, name, |row| {
                    debug!("{row:?}");
                    Ok(())
                })
                .map_err(|err| {
                    error!("cannot update pragma `{name}`=`{value}` - {err:?}");
                    InitializationError
                })
            } else {
                debug!("Setting PRAGMA {name}={value}");
                conn.pragma_update(None, name, value).map_err(|err| {
                    error!("cannot update pragma `{name}`=`{value}` - {err:?}");
                    InitializationError
                })
            }
        }

        let mut conn = Connection::open_with_flags(path, OpenFlags::default()).map_err(|err| {
            error!("cannot open the connection - {err:?}");
            InitializationError
        })?;

        for [pragma_name, default_value] in PRAGMA {
            let pragma_value = pragma_override
                .remove(pragma_name)
                .unwrap_or_else(|| default_value.to_string());
            pragma_update(&conn, pragma_name, &pragma_value)?;
        }
        // drain the rest overrides
        for (pragma_name, pragma_value) in pragma_override.drain() {
            pragma_update(&conn, &pragma_name, &pragma_value)?;
        }

        embedded::migrations::runner()
            .run(&mut conn)
            .map_err(|err| {
                error!("Cannot run migrations - {err:?}");
                InitializationError
            })?;
        Ok(conn)
    }

    fn connection_rpc(
        mut conn: Connection,
        shutdown_requested: &AtomicBool,
        shutdown_finished: &AtomicBool,
        mut command_rx: mpsc::Receiver<ThreadCommand>,
        metrics_threshold: Option<Duration>,
    ) {
        let mut histograms = Histograms::new(metrics_threshold);
        while Self::tick(
            &mut conn,
            shutdown_requested,
            &mut command_rx,
            &mut histograms,
        )
        .is_ok()
        {
            // Loop until shutdown is set to true.
        }
        debug!("Closing command thread");
        shutdown_finished.store(true, Ordering::Release);
    }

    fn tick(
        conn: &mut Connection,
        shutdown_requested: &AtomicBool,
        command_rx: &mut mpsc::Receiver<ThreadCommand>,
        histograms: &mut Histograms,
    ) -> Result<(), ShutdownRequested> {
        #[derive(Clone, Copy, PartialEq, Eq)]
        enum ApplyOrSkip {
            Apply,
            Skip, // LTX apply failed previously
        }
        let mut ltx_list: Vec<(LogicalTx, ApplyOrSkip)> = Vec::new();
        // perf: Wait for first logical tx with high priority.
        loop {
            let ltx = match command_rx.blocking_recv() {
                Some(ThreadCommand::LogicalTx(ltx)) => ltx,
                Some(ThreadCommand::Shutdown) => {
                    debug!("Shutdown message received");
                    return Err(ShutdownRequested);
                }
                None => {
                    debug!("command_rx was closed");
                    return Err(ShutdownRequested);
                }
            };
            let prio = ltx.priority;
            ltx_list.push((ltx, ApplyOrSkip::Apply));
            if prio == LtxPriority::High {
                break;
            }
        }
        // Exactly one High prio LTX is in `ltx_list`.

        let all_fns_start = std::time::Instant::now();

        // Add all remaining LTXes that were queued after the first High-prio.
        while let Ok(more) = command_rx.try_recv() {
            let ltx = match more {
                ThreadCommand::Shutdown => {
                    debug!("Shutdown message received");
                    // ptx gets rolled back on drop
                    // phytx_flush_receiver drop is converted to `RusqliteError::Close` error
                    return Err(ShutdownRequested);
                }
                ThreadCommand::LogicalTx(ltx) => ltx,
            };
            ltx_list.push((ltx, ApplyOrSkip::Apply));
        }

        struct NeedsRestart;
        type CommitResult = Result<(), CommitError>;
        fn try_apply_all(
            mut ptx: Transaction<'_>,
            ltx_list: &mut [(LogicalTx, ApplyOrSkip)],
            histograms: &mut Histograms,
            all_fns_start: Instant,
        ) -> Result<CommitResult, NeedsRestart> {
            for (ltx, former_res) in ltx_list
                .iter_mut()
                .filter(|(_, former_res)| *former_res == ApplyOrSkip::Apply)
            {
                if let Ok(()) = SqlitePool::ltx_apply_to_phytx(ltx, &mut ptx, histograms) {
                } else {
                    *former_res = ApplyOrSkip::Skip; // the problematic ltx will be skipped in the next iteration.
                    // ptx rollbacks on drop
                    return Err(NeedsRestart);
                }
            }
            // All LTXes were applied or skipped.
            histograms.record_all_fns(all_fns_start.elapsed());
            let now = std::time::Instant::now();
            let commit_result = ptx.commit().map_err(|err| {
                warn!("Cannot commit transaction - {err:?}");
                CommitError(RusqliteError::from(err))
            });
            histograms.record_commit(now.elapsed());
            Ok(commit_result)
        }

        fn apply_all(
            conn: &mut Connection,
            ltx_list: &mut [(LogicalTx, ApplyOrSkip)],
            histograms: &mut Histograms,
            all_fns_start: Instant,
            shutdown_requested: &AtomicBool,
        ) -> Result<CommitResult, ShutdownRequested> {
            // TODO: Investigate SAVEPOINT + RELEASE SAVEPOINT, ROLLBACK savepoint instead.
            loop {
                match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
                    Ok(ptx) => {
                        if let Ok(commit_res) =
                            try_apply_all(ptx, ltx_list, histograms, all_fns_start)
                        {
                            return Ok(commit_res);
                        }
                    }
                    Err(begin_err) => {
                        error!("Cannot open transaction - {begin_err:?}");
                        std::thread::sleep(Duration::from_millis(100));
                        if shutdown_requested.load(Ordering::Acquire) {
                            return Err(ShutdownRequested);
                        }
                    }
                }
            }
        }
        let ok_or_commit_error = apply_all(
            conn,
            &mut ltx_list,
            histograms,
            all_fns_start,
            shutdown_requested,
        )?;

        for (ltx, apply_or_skip) in ltx_list {
            let to_send = match apply_or_skip {
                ApplyOrSkip::Apply => ok_or_commit_error.clone(),
                ApplyOrSkip::Skip => {
                    Ok(()) // tx was aborted, caller will receive error from the LTX fn
                }
            };
            // Ignore the sending result: ThreadCommand producer timed out before awaiting the ack.
            let _ = ltx.phytx_flush_sender.send(to_send);
        }

        histograms.print_if_elapsed();
        Ok(())
    }

    // Returning error means the PhyTx needs to be rolled back, implying LTX contained multiple writes.
    fn ltx_apply_to_phytx(
        ltx: &mut LogicalTx,
        physical_tx: &mut Transaction,
        histograms: &mut Histograms,
    ) -> Result<(), ShouldRollback> {
        let sent_latency = ltx.sent_at.elapsed();
        let started_at = Instant::now();
        let res = (ltx.func)(physical_tx);
        histograms.record_command(sent_latency, ltx.func_name, started_at.elapsed());
        res
    }

    #[instrument(skip_all, name = "sqlite_new")]
    pub async fn new<P: AsRef<Path>>(
        path: P,
        config: SqliteConfig,
    ) -> Result<Self, InitializationError> {
        let path = path.as_ref().to_owned();

        let shutdown_requested = Arc::new(AtomicBool::new(false));
        let shutdown_finished = Arc::new(AtomicBool::new(false));

        let (command_tx, command_rx) = tokio::sync::mpsc::channel(config.queue_capacity);
        info!("Sqlite database location: {path:?}");
        let join_handle = {
            // Initialize the `Connection`.
            let init_task = {
                tokio::task::spawn_blocking(move || {
                    Self::init_thread(&path, config.pragma_override.unwrap_or_default())
                })
                .await
            };
            let conn = match init_task {
                Ok(res) => res?,
                Err(join_err) => {
                    error!("Initialization panic - {join_err:?}");
                    return Err(InitializationError);
                }
            };
            let shutdown_requested = shutdown_requested.clone();
            let shutdown_finished = shutdown_finished.clone();
            // Start the RPC thread.
            std::thread::spawn(move || {
                Self::connection_rpc(
                    conn,
                    &shutdown_requested,
                    &shutdown_finished,
                    command_rx,
                    config.metrics_threshold,
                );
            })
        };
        Ok(SqlitePool(SqlitePoolInner {
            shutdown_requested,
            shutdown_finished,
            command_tx,
            response_subscribers: Arc::default(),
            pending_subscribers: Arc::default(),
            join_handle: Some(Arc::new(join_handle)),
            execution_finished_subscribers: Arc::default(),
        }))
    }

    /// Invokes the provided function wrapping a new [`rusqlite::Transaction`] that is committed automatically.
    async fn transaction<F, T, E>(
        &self,
        mut func: F,
        tx_type: TxType,
        func_name: &'static str,
    ) -> Result<T, E>
    where
        F: FnMut(&mut rusqlite::Transaction) -> Result<T, E> + Send + 'static,
        T: Send + 'static,
        E: From<RusqliteError> + Send + 'static,
    {
        let fn_res: Arc<std::sync::Mutex<Option<_>>> = Arc::default();
        let (phytx_flush_sender, phytx_flush_receiver) = oneshot::channel();
        let current_span = Span::current();
        let thread_command_func = {
            let fn_res = fn_res.clone();
            ThreadCommand::LogicalTx(LogicalTx {
                func: Box::new(move |tx| {
                    let _guard = current_span.enter();
                    let func_res = func(tx);
                    let res = if func_res.is_ok() {
                        Ok(())
                    } else {
                        Err(ShouldRollback)
                    };
                    // save result to be sent to the caller
                    *fn_res.lock().unwrap() = Some(func_res);
                    match tx_type {
                        TxType::MultipleWrites => res, // Rollback, skipping this on LTX replay.
                        TxType::Other => Ok(()), // Failed single write should not rollback the PhyTx
                    }
                }),
                sent_at: Instant::now(),
                func_name,
                phytx_flush_sender,
                priority: LtxPriority::High,
            })
        };
        self.0
            .command_tx
            .send(thread_command_func)
            .await
            .map_err(|_send_err| RusqliteError::Close)?;

        // Wait for commit / rollbeck, then get the retval from the mutex.
        match phytx_flush_receiver.await {
            Ok(Ok(())) => {
                let mut guard = fn_res.lock().unwrap();
                std::mem::take(guard.deref_mut()).expect("ltx must have been run at least once")
            }
            Ok(Err(CommitError(rusqlite_err))) => Err(E::from(rusqlite_err)),
            Err(_) => Err(E::from(RusqliteError::Close)),
        }
    }

    /// Invokes the provided function wrapping a new [`rusqlite::Transaction`] that is committed automatically.
    async fn transaction_fire_forget<F, T, E>(&self, mut func: F, func_name: &'static str)
    where
        F: FnMut(&mut rusqlite::Transaction) -> Result<T, E> + Send + 'static,
        T: Send + 'static + Default,
        E: From<RusqliteError> + Send + 'static,
    {
        let (commit_ack_sender, _commit_ack_receiver) = oneshot::channel(); // Nobody is interested in receiving the result
        let current_span = Span::current();
        let thread_command_func = {
            ThreadCommand::LogicalTx(LogicalTx {
                func: Box::new(move |tx| {
                    let _guard = current_span.enter();
                    let _ = func(tx);
                    Ok(()) // Never rollback
                }),
                sent_at: Instant::now(),
                func_name,
                phytx_flush_sender: commit_ack_sender,
                priority: LtxPriority::Low,
            })
        };
        let _ = self.0.command_tx.send(thread_command_func).await; // Ignore error when the channel is closed.
    }

    fn fetch_created_event(
        conn: &Connection,
        execution_id: &ExecutionId,
    ) -> Result<CreateRequest, DbErrorRead> {
        let mut stmt = conn.prepare(
            "SELECT created_at, json_value FROM t_execution_log WHERE \
            execution_id = :execution_id AND version = 0",
        )?;
        let (created_at, event) = stmt.query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
            },
            |row| {
                let created_at = row.get("created_at")?;
                let event = row
                    .get::<_, JsonWrapper<ExecutionRequest>>("json_value")
                    .map_err(|serde| {
                        error!("cannot deserialize `Created` event: {row:?} - `{serde:?}`");
                        consistency_rusqlite("cannot deserialize `Created` event")
                    })?;
                Ok((created_at, event.0))
            },
        )?;
        if let ExecutionRequest::Created {
            ffqn,
            params,
            parent,
            scheduled_at,
            component_id,
            deployment_id,
            metadata,
            scheduled_by,
        } = event
        {
            Ok(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn,
                params,
                parent,
                scheduled_at,
                component_id,
                deployment_id,
                metadata,
                scheduled_by,
                paused: false,
            })
        } else {
            error!("Row with version=0 must be a `Created` event - {event:?}");
            Err(consistency_db_err("expected `Created` event").into())
        }
    }

    fn check_expected_next_and_appending_version(
        expected_version: &Version,
        appending_version: &Version,
    ) -> Result<(), DbErrorWrite> {
        if *expected_version != *appending_version {
            debug!(
                "Version conflict - expected: {expected_version:?}, appending: {appending_version:?}"
            );
            return Err(DbErrorWrite::NonRetriable(
                DbErrorWriteNonRetriable::VersionConflict {
                    expected: expected_version.clone(),
                    requested: appending_version.clone(),
                },
            ));
        }
        Ok(())
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(execution_id = %req.execution_id))]
    fn create_inner(
        tx: &Transaction,
        req: CreateRequest,
    ) -> Result<(AppendResponse, AppendNotifier), DbErrorWrite> {
        trace!("create_inner");

        let version = Version::default();
        let execution_id = req.execution_id.clone();
        let execution_id_str = execution_id.to_string();
        let ffqn = req.ffqn.clone();
        let created_at = req.created_at;
        let scheduled_at = req.scheduled_at;
        let component_id = req.component_id.clone();
        let deployment_id = req.deployment_id;
        let paused = req.paused;
        let event = ExecutionRequest::from(req);
        let event_ser = serde_json::to_string(&event).map_err(|err| {
            error!("Cannot serialize {event:?} - {err:?}");
            DbErrorWriteNonRetriable::ValidationFailed("parameter serialization error".into())
        })?;
        tx.prepare(
                "INSERT INTO t_execution_log (execution_id, created_at, version, json_value, variant, join_set_id ) \
                VALUES (:execution_id, :created_at, :version, :json_value, :variant, :join_set_id)")
        ?
        .execute(named_params! {
            ":execution_id": &execution_id_str,
            ":created_at": created_at,
            ":version": version.0,
            ":json_value": event_ser,
            ":variant": event.variant(),
            ":join_set_id": event.join_set_id().map(std::string::ToString::to_string),
        })
        ?;
        let pending_at = {
            debug!("Creating with `Pending(`{scheduled_at:?}`)");
            tx.prepare(
                r"
                INSERT INTO t_state (
                    execution_id,
                    is_top_level,
                    corresponding_version,
                    pending_expires_finished,
                    ffqn,
                    state,
                    created_at,
                    component_id_input_digest,
                    component_type,
                    deployment_id,
                    updated_at,
                    first_scheduled_at,
                    intermittent_event_count
                    )
                VALUES (
                    :execution_id,
                    :is_top_level,
                    :corresponding_version,
                    :pending_expires_finished,
                    :ffqn,
                    :state,
                    :created_at,
                    :component_id_input_digest,
                    :component_type,
                    :deployment_id,
                    CURRENT_TIMESTAMP,
                    :first_scheduled_at,
                    0
                    )
                ",
            )? // lifecycle defaults to 'active'
            .execute(named_params! {
                ":execution_id": execution_id.to_string(),
                ":is_top_level": execution_id.is_top_level(),
                ":corresponding_version": version.0,
                ":pending_expires_finished": scheduled_at,
                ":ffqn": ffqn.to_string(),
                ":state": STATE_PENDING_AT,
                ":created_at": created_at,
                ":component_id_input_digest": component_id.component_digest,
                ":component_type": component_id.component_type,
                ":deployment_id": deployment_id.to_string(),
                ":first_scheduled_at": scheduled_at,
            })?;
            AppendNotifier {
                pending_at: if paused {
                    None
                } else {
                    Some(NotifierPendingAt {
                        scheduled_at,
                        ffqn: ffqn.clone(),
                        component_input_digest: component_id.component_digest,
                    })
                },
                execution_finished: None,
                response: None,
            }
        };
        let mut next_version = Version::new(version.0 + 1);
        if paused {
            // Append the Paused event in the same transaction.
            let (v, _) = Self::append(
                tx,
                &execution_id,
                AppendRequest {
                    created_at,
                    event: ExecutionRequest::Paused,
                },
                next_version,
            )?;
            next_version = v;
        }
        Ok((next_version, pending_at))
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %scheduled_at))]
    fn update_state_pending_after_response_appended(
        tx: &Transaction,
        execution_id: &ExecutionId,
        scheduled_at: DateTime<Utc>, // Changing to state PendingAt
        component_input_digest: ComponentDigest,
    ) -> Result<AppendNotifier, DbErrorWrite> {
        debug!("Setting t_state to Pending(`{scheduled_at:?}`) after response appended");
        let mut stmt = tx
            .prepare_cached(
                r"
                UPDATE t_state
                SET
                    pending_expires_finished = :pending_expires_finished,
                    state = :state,
                    updated_at = CURRENT_TIMESTAMP,
                    max_retries = NULL,
                    retry_exp_backoff_millis = NULL,
                    last_lock_version = NULL,

                    join_set_id = NULL,
                    join_set_closing = NULL,

                    result_kind = NULL
                WHERE execution_id = :execution_id
            ",
            )
            .map_err(|err| DbErrorGeneric::Uncategorized {
                reason: err.to_string().into(),
                context: SpanTrace::capture(),
                source: Some(Arc::new(err)),
                loc: Location::caller(),
            })?;
        let updated = stmt
            .execute(named_params! {
                ":execution_id": execution_id,
                ":pending_expires_finished": scheduled_at,
                ":state": STATE_PENDING_AT,
            })
            .map_err(|err| DbErrorGeneric::Uncategorized {
                reason: err.to_string().into(),
                context: SpanTrace::capture(),
                source: Some(Arc::new(err)),
                loc: Location::caller(),
            })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(AppendNotifier {
            pending_at: Some(NotifierPendingAt {
                scheduled_at,
                ffqn: Self::fetch_created_event(tx, execution_id)?.ffqn,
                component_input_digest,
            }),
            execution_finished: None,
            response: None,
        })
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, scheduled_at = %update.scheduled_at, %appending_version))]
    fn update_state_pending_after_event_appended(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
        update: PendingAfterEventUpdate,
    ) -> Result<(AppendResponse, AppendNotifier), DbErrorWrite> {
        let scheduled_at = update.scheduled_at;
        debug!("Setting t_state to Pending(`{scheduled_at:?}`) after event appended");
        let sql = r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    pending_expires_finished = :pending_expires_finished,
                    state = :state,
                    updated_at = CURRENT_TIMESTAMP,
                    intermittent_event_count = intermittent_event_count + :intermittent_delta,

                    max_retries = NULL,
                    retry_exp_backoff_millis = NULL,
                    last_lock_version = NULL,

                    join_set_id = NULL,
                    join_set_closing = NULL,

                    result_kind = NULL
                WHERE execution_id = :execution_id;
            "; // `executor_id` and `run_id` are preserved for lock extension.
        let mut stmt = tx.prepare_cached(sql)?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id.to_string(),
            ":appending_version": appending_version.0,
            ":pending_expires_finished": scheduled_at,
            ":state": STATE_PENDING_AT,
            ":intermittent_delta": i32::from(update.intermittent_failure),
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok((
            appending_version.increment(),
            AppendNotifier {
                pending_at: Some(NotifierPendingAt {
                    scheduled_at,
                    ffqn: Self::fetch_created_event(tx, execution_id)?.ffqn,
                    component_input_digest: update.component_input_digest,
                }),
                execution_finished: None,
                response: None,
            },
        ))
    }

    fn update_state_component_upgrade_finished_success(
        tx: &Transaction,
        execution_id: &ExecutionId,
        component_digest: &ComponentDigest,
        deployment_id: DeploymentId,
        appending_version: &Version,
    ) -> Result<AppendResponse, DbErrorWrite> {
        debug!("Updating t_state to component {component_digest}");
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    updated_at = CURRENT_TIMESTAMP,
                    component_id_input_digest = :component_digest,
                    deployment_id = :deployment_id,
                    incompatible_digest = NULL
                WHERE execution_id = :execution_id;
            ",
        )?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id.to_string(),
            ":appending_version": appending_version.0,
            ":component_digest": component_digest,
            ":deployment_id": deployment_id.to_string(),
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(appending_version.increment())
    }

    fn update_state_component_upgrade_finished_failed(
        tx: &Transaction,
        execution_id: &ExecutionId,
        target_digest: &ComponentDigest,
        appending_version: &Version,
    ) -> Result<AppendResponse, DbErrorWrite> {
        debug!("Marking component {target_digest} incompatible after upgrade failure");
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    updated_at = CURRENT_TIMESTAMP,
                    incompatible_digest = :target_digest
                WHERE execution_id = :execution_id;
            ",
        )?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id.to_string(),
            ":appending_version": appending_version.0,
            ":target_digest": target_digest,
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(appending_version.increment())
    }

    #[expect(clippy::too_many_arguments)]
    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version))]
    fn update_state_locked_get_intermittent_event_count(
        tx: &Transaction,
        execution_id: &ExecutionId,
        deployment_id: Option<DeploymentId>,
        component_digest: Option<&ComponentDigest>,
        executor_id: ExecutorId,
        run_id: RunId,
        lock_expires_at: DateTime<Utc>,
        appending_version: &Version,
        retry_config: ComponentRetryConfig,
    ) -> Result<u32, DbErrorWrite> {
        debug!("Setting t_state to Locked(`{lock_expires_at:?}`)");
        let backoff_millis =
            i64::try_from(retry_config.retry_exp_backoff.as_millis()).map_err(|err| {
                DbErrorGeneric::Uncategorized {
                    reason: "backoff too big".into(),
                    context: SpanTrace::capture(),
                    source: Some(Arc::new(err)),
                    loc: Location::caller(),
                }
            })?; // Keep equal to Postgres' BIGINT = i64
        let execution_id_str = execution_id.to_string();
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    pending_expires_finished = :pending_expires_finished,
                    state = :state,
                    updated_at = CURRENT_TIMESTAMP,
                    deployment_id = COALESCE(:deployment_id, deployment_id),
                    component_id_input_digest = COALESCE(:component_id_input_digest, component_id_input_digest),

                    max_retries = :max_retries,
                    retry_exp_backoff_millis = :retry_exp_backoff_millis,
                    last_lock_version = :appending_version,
                    executor_id = :executor_id,
                    run_id = :run_id,

                    join_set_id = NULL,
                    join_set_closing = NULL,

                    result_kind = NULL
                WHERE execution_id = :execution_id
                AND lifecycle = 'active'
            ",
        )?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
            ":pending_expires_finished": lock_expires_at,
            ":state": STATE_LOCKED,
            ":deployment_id": deployment_id, // no change if `None` due to `coalesce`
            ":component_id_input_digest": component_digest.cloned(), // no change if `None` due to `coalesce`
            ":max_retries": retry_config.max_retries,
            ":retry_exp_backoff_millis": backoff_millis,
            ":executor_id": executor_id.to_string(),
            ":run_id": run_id.to_string(),
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }

        // fetch intermittent event count from the just-inserted row.
        let intermittent_event_count = tx
            .prepare(
                "SELECT intermittent_event_count FROM t_state WHERE execution_id = :execution_id",
            )?
            .query_row(
                named_params! {
                    ":execution_id": execution_id_str,
                },
                |row| {
                    let intermittent_event_count = row.get("intermittent_event_count")?;
                    Ok(intermittent_event_count)
                },
            )?;

        Ok(intermittent_event_count)
    }

    /// Appending [`HistoryEvent::JoinNext`].
    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version))]
    fn update_state_blocked(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
        // BlockedByJoinSet fields
        join_set_id: &JoinSetId,
        lock_expires_at: DateTime<Utc>,
        join_set_closing: bool,
    ) -> Result<
        AppendResponse, // next version
        DbErrorWrite,
    > {
        debug!("Setting t_state to BlockedByJoinSet(`{join_set_id}`)");
        let execution_id_str = execution_id.to_string();
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    pending_expires_finished = :pending_expires_finished,
                    state = :state,
                    updated_at = CURRENT_TIMESTAMP,

                    max_retries = NULL,
                    retry_exp_backoff_millis = NULL,
                    last_lock_version = NULL,

                    join_set_id = :join_set_id,
                    join_set_closing = :join_set_closing,

                    result_kind = NULL
                WHERE execution_id = :execution_id
            ",
        )?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
            ":pending_expires_finished": lock_expires_at,
            ":state": STATE_BLOCKED_BY_JOIN_SET,
            ":join_set_id": join_set_id,
            ":join_set_closing": join_set_closing,
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(appending_version.increment())
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version))]
    fn update_state_finished(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
        // Finished fields
        finished_at: DateTime<Utc>,
        result_kind: PendingStateFinishedResultKind,
    ) -> Result<(), DbErrorWrite> {
        debug!("Setting t_state to Finished");
        let execution_id_str = execution_id.to_string();
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    pending_expires_finished = :pending_expires_finished,
                    state = :state,
                    updated_at = CURRENT_TIMESTAMP,

                    max_retries = NULL,
                    retry_exp_backoff_millis = NULL,
                    last_lock_version = NULL,
                    executor_id = NULL,
                    run_id = NULL,

                    join_set_id = NULL,
                    join_set_closing = NULL,

                    lifecycle = 'active',
                    result_kind = :result_kind
                WHERE execution_id = :execution_id
            ",
        )?;

        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
            ":pending_expires_finished": finished_at,
            ":state": STATE_FINISHED,
            ":result_kind": JsonWrapper(result_kind),
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(())
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version, %is_paused))]
    fn update_state_paused(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
        is_paused: bool,
    ) -> Result<AppendResponse, DbErrorWrite> {
        debug!(
            "Setting t_state to {}",
            if is_paused { "paused" } else { "unpaused" }
        );
        let execution_id_str = execution_id.to_string();
        let lifecycle = if is_paused {
            LIFECYCLE_PAUSED
        } else {
            LIFECYCLE_ACTIVE
        };
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    lifecycle = :lifecycle,
                    updated_at = CURRENT_TIMESTAMP
                WHERE execution_id = :execution_id
            ",
        )?;

        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
            ":lifecycle": lifecycle,
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(appending_version.increment())
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version))]
    fn update_state_cancelling(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
    ) -> Result<AppendResponse, DbErrorWrite> {
        debug!("Setting t_state lifecycle to cancelling");
        let execution_id_str = execution_id.to_string();
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    lifecycle = :lifecycle,
                    updated_at = CURRENT_TIMESTAMP
                WHERE execution_id = :execution_id
            ",
        )?;

        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
            ":lifecycle": LIFECYCLE_CANCELLING,
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(appending_version.increment())
    }

    // Upon appending new event to t_execution_log, copy the previous t_state with changed appending_version and created_at.
    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %appending_version))]
    fn bump_state_next_version(
        tx: &Transaction,
        execution_id: &ExecutionId,
        appending_version: &Version,
        delay_req: Option<DelayReq>,
    ) -> Result<AppendResponse /* next version */, DbErrorWrite> {
        debug!("update_index_version");
        let execution_id_str = execution_id.to_string();
        let mut stmt = tx.prepare_cached(
            r"
                UPDATE t_state
                SET
                    corresponding_version = :appending_version,
                    updated_at = CURRENT_TIMESTAMP
                WHERE execution_id = :execution_id
            ",
        )?;
        let updated = stmt.execute(named_params! {
            ":execution_id": execution_id_str,
            ":appending_version": appending_version.0,
        })?;
        if updated != 1 {
            return Err(DbErrorWrite::NotFound);
        }
        if let Some(DelayReq {
            join_set_id,
            delay_id,
            expires_at,
            paused,
        }) = delay_req
        {
            debug!("Inserting delay to `t_delay`");
            let mut stmt = tx.prepare_cached(
                "INSERT INTO t_delay (execution_id, join_set_id, delay_id, expires_at, is_paused) \
                VALUES \
                (:execution_id, :join_set_id, :delay_id, :expires_at, :is_paused)",
            )?;
            stmt.execute(named_params! {
                ":execution_id": execution_id_str,
                ":join_set_id": join_set_id.to_string(),
                ":delay_id": delay_id.to_string(),
                ":expires_at": expires_at,
                ":is_paused": paused,
            })?;
        }
        Ok(appending_version.increment())
    }

    fn get_combined_state(
        tx: &Transaction,
        execution_id: &ExecutionId,
    ) -> Result<CombinedState, DbErrorRead> {
        let mut stmt = tx.prepare(
            r"
                SELECT
                    created_at, first_scheduled_at,
                    state, ffqn, component_id_input_digest, component_type, deployment_id,
                    corresponding_version, pending_expires_finished,
                    last_lock_version, executor_id, run_id,
                    join_set_id, join_set_closing,
                    result_kind, lifecycle
                    FROM t_state
                WHERE
                    execution_id = :execution_id
                ",
        )?;
        stmt.query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
            },
            |row| {
                CombinedState::new(
                    CombinedStateDTO {
                        execution_id: execution_id.clone(),
                        created_at: row.get("created_at")?,
                        first_scheduled_at: row.get("first_scheduled_at")?,
                        component_digest: row.get("component_id_input_digest")?,
                        component_type: row.get("component_type")?,
                        deployment_id: row.get("deployment_id")?,
                        state: row.get("state")?,
                        ffqn: row.get("ffqn")?,
                        pending_expires_finished: row
                            .get::<_, DateTime<Utc>>("pending_expires_finished")?,
                        last_lock_version: row
                            .get::<_, Option<VersionType>>("last_lock_version")?
                            .map(Version::new),
                        executor_id: row.get::<_, Option<ExecutorId>>("executor_id")?,
                        run_id: row.get::<_, Option<RunId>>("run_id")?,
                        join_set_id: row.get::<_, Option<JoinSetId>>("join_set_id")?,
                        join_set_closing: row.get::<_, Option<bool>>("join_set_closing")?,
                        result_kind: row
                            .get::<_, Option<JsonWrapper<PendingStateFinishedResultKind>>>(
                                "result_kind",
                            )?
                            .map(|wrapper| wrapper.0),
                        lifecycle: Lifecycle::from_column(&row.get::<_, String>("lifecycle")?)
                            .ok_or_else(|| consistency_rusqlite("invalid t_state.lifecycle"))?,
                    },
                    Version::new(row.get("corresponding_version")?),
                )
                .map_err(|e| from_generic_error(&e))
            },
        )
        .map_err(DbErrorRead::from)
    }

    fn list_executions(
        read_tx: &Transaction,
        filter: &ListExecutionsFilter,
        pagination: &ExecutionListPagination,
    ) -> Result<Vec<ExecutionWithState>, RusqliteError> {
        #[derive(Debug)]
        struct StatementModifier<'a> {
            where_vec: Vec<String>,
            params: Vec<(&'static str, ToSqlOutput<'a>)>,
            limit: u32,
            limit_desc: bool,
        }

        fn paginate<'a, T: Clone + rusqlite::ToSql + 'static>(
            pagination: &'a Pagination<Option<T>>,
            column: &str,
            filter: &ListExecutionsFilter,
        ) -> Result<StatementModifier<'a>, RusqliteError> {
            let mut where_vec: Vec<String> = vec![];
            let mut params: Vec<(&'static str, ToSqlOutput<'a>)> = vec![];
            let limit = pagination.length();
            let limit_desc = pagination.is_desc();
            match pagination {
                Pagination::NewerThan {
                    cursor: Some(cursor),
                    ..
                }
                | Pagination::OlderThan {
                    cursor: Some(cursor),
                    ..
                } => {
                    where_vec.push(format!("{column} {rel} :cursor", rel = pagination.rel()));
                    let cursor = cursor.to_sql().map_err(|err| {
                        error!("Possible program error - cannot convert cursor to sql - {err:?}");
                        RusqliteError::Generic {
                            reason: "cannot convert cursor to sql".into(),
                            context: SpanTrace::capture(),
                            source: Some(Arc::new(err)),
                            loc: Location::caller(),
                        }
                    })?;
                    params.push((":cursor", cursor));
                }
                _ => {}
            }
            if !filter.show_derived {
                where_vec.push("is_top_level=true".to_string());
            }
            Ok(StatementModifier {
                where_vec,
                params,
                limit: u32::from(limit),
                limit_desc,
            })
        }

        let mut statement_mod = match pagination {
            ExecutionListPagination::CreatedBy(pagination) => {
                paginate(pagination, "created_at", filter)?
            }
            ExecutionListPagination::ExecutionId(pagination) => {
                paginate(pagination, "execution_id", filter)?
            }
        };
        if let Some(function_name_filter) = &filter.function_name_filter {
            let ffqn = ToSqlOutput::Owned(Value::from(function_name_filter.like_pattern()));
            statement_mod.where_vec.push("ffqn LIKE :ffqn".to_string());
            statement_mod.params.push((":ffqn", ffqn));
        }
        let like = |value: &str| format!("{value}%");

        if filter.hide_finished {
            statement_mod
                .where_vec
                .push(format!("state != '{STATE_FINISHED}'"));
        }
        let prefix_temporary;
        if let Some(prefix) = &filter.execution_id_prefix {
            statement_mod
                .where_vec
                .push("execution_id LIKE :prefix".to_string());
            prefix_temporary = like(prefix);
            statement_mod.params.push((
                ":prefix",
                prefix_temporary
                    .to_sql()
                    .expect("string conversion never fails"),
            ));
        }

        let component_digest_temporary;
        if let Some(componnet_digest) = &filter.component_digest {
            statement_mod
                .where_vec
                .push("component_id_input_digest = :component_digest".to_string());
            component_digest_temporary = componnet_digest.clone();
            let component_digest_sql = component_digest_temporary
                .to_sql()
                .expect("InputContentDigest conversion never fails");
            statement_mod
                .params
                .push((":component_digest", component_digest_sql));
        }

        let deployment_id_temporary;
        if let Some(deployment_id) = filter.deployment_id {
            statement_mod
                .where_vec
                .push("deployment_id = :deployment_id".to_string());
            deployment_id_temporary = deployment_id;
            let deployment_id = deployment_id_temporary
                .to_sql()
                .expect("DeploymentId conversion never fails");
            statement_mod.params.push((":deployment_id", deployment_id));
        }

        let state_filter_now_temporary;
        if !filter.state_filters.is_empty() {
            let conditions: Vec<String> = filter
                .state_filters
                .iter()
                .map(|state_filter| state_filter_to_sql(state_filter, ":state_filter_now", ""))
                .collect();
            statement_mod
                .where_vec
                .push(format!("({})", conditions.join(" OR ")));
            if let Some(now) = state_filters_now(&filter.state_filters) {
                state_filter_now_temporary = now;
                statement_mod.params.push((
                    ":state_filter_now",
                    state_filter_now_temporary
                        .to_sql()
                        .expect("DateTime conversion never fails"),
                ));
            }
        }

        let where_str = if statement_mod.where_vec.is_empty() {
            String::new()
        } else {
            format!("WHERE {}", statement_mod.where_vec.join(" AND "))
        };

        // Inner query: fetch rows with cursor-based ordering
        // Outer query: always return results in descending order
        let (inner_order, outer_order) = if statement_mod.limit_desc {
            ("DESC", "")
        } else {
            ("", "DESC")
        };

        let inner_sql = format!(
            r"SELECT created_at, first_scheduled_at, component_id_input_digest, component_type, deployment_id,
            state, execution_id, ffqn, corresponding_version, pending_expires_finished,
            last_lock_version, executor_id, run_id,
            join_set_id, join_set_closing,
            result_kind, lifecycle
            FROM t_state {where_str} ORDER BY created_at {inner_order} LIMIT {limit}",
            limit = statement_mod.limit,
        );

        let sql = if outer_order.is_empty() {
            inner_sql
        } else {
            format!("SELECT * FROM ({inner_sql}) AS sub ORDER BY created_at {outer_order}")
        };
        let vec: Vec<_> = read_tx
            .prepare(&sql)?
            .query_map::<_, &[(&'static str, ToSqlOutput)], _>(
                statement_mod
                    .params
                    .into_iter()
                    .collect::<Vec<_>>()
                    .as_ref(),
                |row| {
                    let combined_state = CombinedState::new(
                        CombinedStateDTO {
                            execution_id: row.get("execution_id")?,
                            created_at: row.get("created_at")?,
                            first_scheduled_at: row.get("first_scheduled_at")?,
                            component_digest: row.get("component_id_input_digest")?,
                            component_type: row.get("component_type")?,
                            deployment_id: row.get("deployment_id")?,
                            state: row.get("state")?,
                            ffqn: row.get("ffqn")?,
                            pending_expires_finished: row.get("pending_expires_finished")?,
                            executor_id: row.get::<_, Option<ExecutorId>>("executor_id")?,

                            last_lock_version: row
                                .get::<_, Option<VersionType>>("last_lock_version")?
                                .map(Version::new),
                            run_id: row.get::<_, Option<RunId>>("run_id")?,
                            join_set_id: row.get::<_, Option<JoinSetId>>("join_set_id")?,
                            join_set_closing: row.get::<_, Option<bool>>("join_set_closing")?,
                            result_kind: row
                                .get::<_, Option<JsonWrapper<PendingStateFinishedResultKind>>>(
                                    "result_kind",
                                )?
                                .map(|wrapper| wrapper.0),
                            lifecycle: Lifecycle::from_column(&row.get::<_, String>("lifecycle")?)
                                .ok_or_else(|| consistency_rusqlite("invalid t_state.lifecycle"))?,
                        },
                        Version::new(row.get("corresponding_version")?),
                    )
                    .map_err(|e| from_generic_error(&e))?;
                    Ok(combined_state.execution_with_state)
                },
            )?
            .collect::<Vec<Result<_, _>>>()
            .into_iter()
            .filter_map(|row| match row {
                Ok(row) => Some(row),
                Err(err) => {
                    warn!("Skipping row - {err:?}");
                    None
                }
            })
            .collect();

        Ok(vec)
    }

    fn list_responses(
        tx: &Transaction,
        execution_id: &ExecutionId,
        pagination: Option<Pagination<u32>>,
    ) -> Result<Vec<ResponseWithCursor>, DbErrorRead> {
        // TODO: Add test
        let mut params: Vec<(&'static str, Box<dyn rusqlite::ToSql>)> = vec![];
        let mut sql = "SELECT \
            r.id, r.seq, r.created_at, r.join_set_id,  r.delay_id, r.delay_success, r.child_execution_id, r.finished_version, l.json_value \
            FROM t_join_set_response r LEFT OUTER JOIN t_execution_log l ON r.child_execution_id = l.execution_id \
            WHERE \
            r.execution_id = :execution_id \
            AND ( r.finished_version = l.version OR r.child_execution_id IS NULL ) \
            "
        .to_string();
        let limit = match &pagination {
            Some(
                pagination @ (Pagination::NewerThan { cursor, .. }
                | Pagination::OlderThan { cursor, .. }),
            ) => {
                params.push((":cursor", Box::new(cursor)));
                write!(sql, " AND r.seq {rel} :cursor", rel = pagination.rel()).unwrap();
                Some(pagination.length())
            }
            None => None,
        };
        sql.push_str(" ORDER BY seq");
        let is_desc = pagination.as_ref().is_some_and(Pagination::is_desc);
        if is_desc {
            sql.push_str(" DESC");
        }
        if let Some(limit) = limit {
            write!(sql, " LIMIT {limit}").unwrap();
        }
        // Re-order to ascending for consistent oldest-to-newest results
        if is_desc {
            sql = format!("SELECT * FROM ({sql}) ORDER BY seq ASC");
        }
        params.push((":execution_id", Box::new(execution_id.to_string())));
        tx.prepare(&sql)?
            .query_map::<_, &[(&'static str, &dyn ToSql)], _>(
                params
                    .iter()
                    .map(|(key, value)| (*key, value.as_ref()))
                    .collect::<Vec<_>>()
                    .as_ref(),
                Self::parse_response_with_cursor,
            )?
            .collect::<Result<Vec<_>, rusqlite::Error>>()
            .map_err(DbErrorRead::from)
    }

    fn parse_response_with_cursor(
        row: &rusqlite::Row<'_>,
    ) -> Result<ResponseWithCursor, rusqlite::Error> {
        let id: i64 = row.get("id")?;
        let seq: u32 = row.get("seq")?;
        let created_at: DateTime<Utc> = row.get("created_at")?;
        let join_set_id = row.get::<_, JoinSetId>("join_set_id")?;
        let event = match (
            row.get::<_, Option<DelayId>>("delay_id")?,
            row.get::<_, Option<bool>>("delay_success")?,
            row.get::<_, Option<ExecutionIdDerived>>("child_execution_id")?,
            row.get::<_, Option<VersionType>>("finished_version")?,
            row.get::<_, Option<JsonWrapper<ExecutionRequest>>>("json_value")?,
        ) {
            (Some(delay_id), Some(delay_success), None, None, None) => {
                JoinSetResponse::DelayFinished {
                    delay_id,
                    result: delay_success.then_some(()).ok_or(()),
                }
            }
            (
                None,
                None,
                Some(child_execution_id),
                Some(finished_version),
                Some(JsonWrapper(ExecutionRequest::Finished { retval: result, .. })),
            ) => JoinSetResponse::ChildExecutionFinished {
                child_execution_id,
                finished_version: Version(finished_version),
                result,
            },
            (delay, delay_success, child, finished, result) => {
                error!(
                    "Invalid row in t_join_set_response {id} - {delay:?} {delay_success:?} {child:?} {finished:?} {:?}",
                    result.map(|it| it.0)
                );
                return Err(consistency_rusqlite("invalid row in t_join_set_response"));
            }
        };
        Ok(ResponseWithCursor {
            cursor: ResponseCursor(seq),
            event: JoinSetResponseEventOuter {
                event: JoinSetResponseEvent { join_set_id, event },
                created_at,
            },
        })
    }

    /// `component_id` is used to construct the `Locked` event.
    /// If `update_component_digest_and_deployment_id` is set, execution state (`t_state`) will be updated to the component's digest and deployment id.
    /// This should be true in all cases except for `lock_pending_by_ffqns_auto`, where
    /// the digest in `t_state` is only updated after a successful execution upgrade.
    #[instrument(level = Level::TRACE, skip(tx))]
    #[expect(clippy::too_many_arguments)]
    fn lock_single_execution(
        tx: &Transaction,
        created_at: DateTime<Utc>,
        component_id: &ComponentId,
        update_component_digest_and_deployment_id: bool,
        deployment_id: DeploymentId,
        execution_id: &ExecutionId,
        run_id: RunId,
        appending_version: &Version,
        executor_id: ExecutorId,
        lock_expires_at: DateTime<Utc>,
        retry_config: ComponentRetryConfig,
    ) -> Result<LockedExecution, DbErrorWrite> {
        trace!("lock_single_execution");
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        let context_component_digest = if update_component_digest_and_deployment_id {
            component_id.component_digest.clone()
        } else {
            combined_state.execution_with_state.component_digest.clone()
        };
        combined_state
            .execution_with_state
            .pending_state
            .can_append_lock(created_at, executor_id, run_id, lock_expires_at)?;
        let expected_version = combined_state.get_next_version_assert_not_finished();
        Self::check_expected_next_and_appending_version(&expected_version, appending_version)?;

        // Append to `execution_log` table.
        let locked_event = Locked {
            component_id: component_id.clone(),
            deployment_id,
            executor_id,
            lock_expires_at,
            run_id,
            retry_config,
        };
        let event = ExecutionRequest::Locked(locked_event.clone());
        let event_ser = serde_json::to_string(&event).map_err(|err| {
            warn!("Cannot serialize {event:?} - {err:?}");
            DbErrorWriteNonRetriable::ValidationFailed("parameter serialization error".into())
        })?;
        let mut stmt = tx
            .prepare_cached(
                "INSERT INTO t_execution_log \
            (execution_id, created_at, json_value, version, variant) \
            VALUES \
            (:execution_id, :created_at, :json_value, :version, :variant)",
            )
            .map_err(|err| DbErrorGeneric::Uncategorized {
                reason: err.to_string().into(),
                context: SpanTrace::capture(),
                source: Some(Arc::new(err)),
                loc: Location::caller(),
            })?;
        stmt.execute(named_params! {
            ":execution_id": execution_id.to_string(),
            ":created_at": created_at,
            ":json_value": event_ser,
            ":version": appending_version.0,
            ":variant": event.variant(),
        })
        .map_err(|err| {
            DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState {
                reason: "cannot lock".into(),
                context: SpanTrace::capture(),
                source: Some(Arc::new(err)),
                loc: Location::caller(),
            })
        })?;

        let responses = Self::list_responses(tx, execution_id, None)?;
        trace!("Responses: {responses:?}");

        // Update `t_state`
        let intermittent_event_count = Self::update_state_locked_get_intermittent_event_count(
            tx,
            execution_id,
            update_component_digest_and_deployment_id.then_some(deployment_id),
            update_component_digest_and_deployment_id.then_some(&component_id.component_digest),
            executor_id,
            run_id,
            lock_expires_at,
            appending_version,
            retry_config,
        )?;
        // Fetch event_history and `Created` event to construct the response.
        let mut events = tx
            .prepare(
                "SELECT json_value, version FROM t_execution_log WHERE \
                execution_id = :execution_id AND (variant = :variant1 OR variant = :variant2) \
                ORDER BY version",
            )?
            .query_map(
                named_params! {
                    ":execution_id": execution_id.to_string(),
                    ":variant1": DUMMY_CREATED.variant(),
                    ":variant2": DUMMY_HISTORY_EVENT.variant(),
                },
                |row| {
                    let created_at_fake = DateTime::from_timestamp_nanos(0); // not used, only the inner event and version
                    let event = row
                        .get::<_, JsonWrapper<ExecutionRequest>>("json_value")
                        .map_err(|serde| {
                            error!("Cannot deserialize {row:?} - {serde:?}");
                            consistency_rusqlite("cannot deserialize event")
                        })?
                        .0;
                    let version = Version(row.get("version")?);

                    Ok(ExecutionEvent {
                        created_at: created_at_fake,
                        event,
                        backtrace_id: None,
                        version,
                    })
                },
            )?
            .collect::<Result<Vec<_>, _>>()?
            .into_iter()
            .collect::<VecDeque<_>>();
        let Some(ExecutionRequest::Created {
            ffqn,
            params,
            parent,
            metadata,
            ..
        }) = events.pop_front().map(|outer| outer.event)
        else {
            return Err(consistency_db_err("execution log must contain `Created` event").into());
        };

        let event_history = events
            .into_iter()
            .map(|ExecutionEvent { event, version, .. }| {
                if let ExecutionRequest::HistoryEvent { event } = event {
                    Ok((event, version))
                } else {
                    Err(consistency_db_err(
                        "rows can only contain `Created` and `HistoryEvent` event kinds",
                    ))
                }
            })
            .collect::<Result<Vec<_>, _>>()?;

        Ok(LockedExecution {
            execution_id: execution_id.clone(),
            metadata,
            component_digest: context_component_digest,
            next_version: appending_version.increment(),
            ffqn,
            params,
            event_history,
            responses,
            parent,
            intermittent_event_count,
            locked_event,
        })
    }

    fn count_join_next(
        tx: &Transaction,
        execution_id: &ExecutionId,
        join_set_id: &JoinSetId,
    ) -> Result<u32, DbErrorRead> {
        let mut stmt = tx.prepare(
            "SELECT COUNT(*) as count FROM t_execution_log WHERE execution_id = :execution_id AND join_set_id = :join_set_id \
            AND history_event_type = :join_next",
        )?;
        Ok(stmt.query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
                ":join_set_id": join_set_id.to_string(),
                ":join_next": HISTORY_EVENT_TYPE_JOIN_NEXT,
            },
            |row| row.get::<_, u32>("count"),
        )?)
    }

    fn nth_response(
        tx: &Transaction,
        execution_id: &ExecutionId,
        join_set_id: &JoinSetId,
        skip_rows: u32,
    ) -> Result<Option<ResponseWithCursor>, DbErrorRead> {
        // TODO: Add test
        tx
            .prepare(
                "SELECT r.id, r.seq, r.created_at, r.join_set_id, \
                    r.delay_id, r.delay_success, \
                    r.child_execution_id, r.finished_version, l.json_value \
                    FROM t_join_set_response r LEFT OUTER JOIN t_execution_log l ON r.child_execution_id = l.execution_id \
                    WHERE \
                    r.execution_id = :execution_id AND r.join_set_id = :join_set_id AND \
                    (
                    r.finished_version = l.version \
                    OR \
                    r.child_execution_id IS NULL \
                    ) \
                    ORDER BY seq \
                    LIMIT 1 OFFSET :offset",
            )
            ?
            .query_row(
                named_params! {
                    ":execution_id": execution_id.to_string(),
                    ":join_set_id": join_set_id.to_string(),
                    ":offset": skip_rows,
                },
                Self::parse_response_with_cursor,
            )
            .optional()
            .map_err(DbErrorRead::from)
    }

    #[instrument(level = Level::TRACE, skip_all, fields(%execution_id, %appending_version))]
    #[expect(clippy::needless_return)]
    fn append(
        tx: &Transaction,
        execution_id: &ExecutionId,
        req: AppendRequest,
        appending_version: Version,
    ) -> Result<(AppendResponse, AppendNotifier), DbErrorWrite> {
        if matches!(req.event, ExecutionRequest::Created { .. }) {
            return Err(DbErrorWrite::NonRetriable(
                DbErrorWriteNonRetriable::ValidationFailed(
                    "cannot append `Created` event - use `create` instead".into(),
                ),
            ));
        }
        if let AppendRequest {
            event:
                ExecutionRequest::Locked(Locked {
                    component_id,
                    deployment_id,
                    executor_id,
                    run_id,
                    lock_expires_at,
                    retry_config,
                }),
            created_at,
        } = req
        {
            return Self::lock_single_execution(
                tx,
                created_at,
                &component_id,
                true,
                deployment_id,
                execution_id,
                run_id,
                &appending_version,
                executor_id,
                lock_expires_at,
                retry_config,
            )
            .map(|locked_execution| (locked_execution.next_version, AppendNotifier::default()));
        }

        let combined_state = Self::get_combined_state(tx, execution_id)?;
        if combined_state
            .execution_with_state
            .pending_state
            .is_finished()
        {
            debug!("Execution is already finished");
            return Err(DbErrorWrite::NonRetriable(
                DbErrorWriteNonRetriable::AlreadyFinished,
            ));
        }

        Self::check_expected_next_and_appending_version(
            &combined_state.get_next_version_assert_not_finished(),
            &appending_version,
        )?;
        let event_ser = serde_json::to_string(&req.event).map_err(|err| {
            error!("Cannot serialize {:?} - {err:?}", req.event);
            DbErrorWriteNonRetriable::ValidationFailed("parameter serialization error".into())
        })?;

        let mut stmt = tx.prepare(
                    "INSERT INTO t_execution_log (execution_id, created_at, json_value, version, variant, join_set_id) \
                    VALUES (:execution_id, :created_at, :json_value, :version, :variant, :join_set_id)")
                    ?;
        stmt.execute(named_params! {
            ":execution_id": execution_id.to_string(),
            ":created_at": req.created_at,
            ":json_value": event_ser,
            ":version": appending_version.0,
            ":variant": req.event.variant(),
            ":join_set_id": req.event.join_set_id().map(std::string::ToString::to_string),
        })?;
        // Calculate current pending state

        match &req.event {
            ExecutionRequest::Created { .. } => {
                unreachable!("handled in the caller")
            }

            ExecutionRequest::Locked { .. } => {
                unreachable!("handled above")
            }

            ExecutionRequest::TemporarilyFailed {
                backoff_expires_at, ..
            }
            | ExecutionRequest::TemporarilyTimedOut {
                backoff_expires_at, ..
            } => {
                let (next_version, notifier) = Self::update_state_pending_after_event_appended(
                    tx,
                    execution_id,
                    &appending_version,
                    PendingAfterEventUpdate {
                        scheduled_at: *backoff_expires_at,
                        intermittent_failure: true,
                        component_input_digest: combined_state
                            .execution_with_state
                            .component_digest,
                    },
                )?;
                return Ok((next_version, notifier));
            }

            ExecutionRequest::Unlocked(unlocked) => {
                match &combined_state.execution_with_state.pending_state {
                    PendingState::PendingAt(_) => {
                        return Err(DbErrorWrite::NonRetriable(
                            DbErrorWriteNonRetriable::UnlockedCannotBeAppended("pending"),
                        ));
                    }
                    PendingState::Locked(_) => {
                        let (next_version, notifier) =
                            Self::update_state_pending_after_event_appended(
                                tx,
                                execution_id,
                                &appending_version,
                                PendingAfterEventUpdate {
                                    scheduled_at: unlocked.unlocked_at,
                                    intermittent_failure: false,
                                    component_input_digest: combined_state
                                        .execution_with_state
                                        .component_digest,
                                },
                            )?;
                        return Ok((next_version, notifier));
                    }
                    PendingState::BlockedByJoinSet(_) => {
                        return Err(DbErrorWrite::NonRetriable(
                            DbErrorWriteNonRetriable::UnlockedCannotBeAppended("blocked"),
                        ));
                    }
                    PendingState::Paused(_) => {
                        return Err(DbErrorWrite::NonRetriable(
                            DbErrorWriteNonRetriable::UnlockedCannotBeAppended("paused"),
                        ));
                    }
                    PendingState::Cancelling(_) => {
                        return Err(DbErrorWrite::NonRetriable(
                            DbErrorWriteNonRetriable::UnlockedCannotBeAppended("cancelling"),
                        ));
                    }
                    PendingState::Finished(_) => {
                        unreachable!("handled above");
                    }
                }
            }

            ExecutionRequest::ComponentUpgradeFinished {
                component_digest,
                deployment_id,
                outcome,
            } => {
                let next_version = match outcome {
                    ComponentUpgradeOutcome::Success { .. } => {
                        Self::update_state_component_upgrade_finished_success(
                            tx,
                            execution_id,
                            component_digest,
                            *deployment_id,
                            &appending_version,
                        )?
                    }
                    ComponentUpgradeOutcome::Failed { .. } => {
                        Self::update_state_component_upgrade_finished_failed(
                            tx,
                            execution_id,
                            component_digest,
                            &appending_version,
                        )?
                    }
                };
                return Ok((next_version, AppendNotifier::default()));
            }

            ExecutionRequest::Paused => {
                match &combined_state.execution_with_state.pending_state {
                    PendingState::Finished { .. } => {
                        unreachable!("handled above");
                    }
                    PendingState::Locked(..) => {
                        return Err(DbErrorWriteNonRetriable::IllegalState {
                            reason:
                                "cannot append Paused event when execution is locked; use pause_execution"
                                    .into(),
                            context: SpanTrace::capture(),
                            source: None,
                            loc: Location::caller(),
                        }
                        .into());
                    }
                    PendingState::Paused(..) => {
                        return Err(DbErrorWriteNonRetriable::IllegalState {
                            reason: "cannot pause, execution is already paused".into(),
                            context: SpanTrace::capture(),
                            source: None,
                            loc: Location::caller(),
                        }
                        .into());
                    }
                    PendingState::Cancelling(..) => {
                        return Err(DbErrorWriteNonRetriable::IllegalState {
                            reason: "cannot pause, execution is cancelling".into(),
                            context: SpanTrace::capture(),
                            source: None,
                            loc: Location::caller(),
                        }
                        .into());
                    }
                    _ => {}
                }
                let next_version =
                    Self::update_state_paused(tx, execution_id, &appending_version, true)?;
                return Ok((next_version, AppendNotifier::default()));
            }

            ExecutionRequest::Unpaused => {
                if !combined_state
                    .execution_with_state
                    .pending_state
                    .is_paused()
                {
                    return Err(DbErrorWriteNonRetriable::IllegalState {
                        reason: "cannot unpause, execution is not paused".into(),
                        context: SpanTrace::capture(),
                        source: None,
                        loc: Location::caller(),
                    }
                    .into());
                }
                let next_version =
                    Self::update_state_paused(tx, execution_id, &appending_version, false)?;
                return Ok((next_version, AppendNotifier::default()));
            }

            ExecutionRequest::CancellationRequested => {
                match &combined_state.execution_with_state.pending_state {
                    PendingState::Finished { .. } => {
                        unreachable!("handled above");
                    }
                    PendingState::Cancelling(..) => {
                        return Err(DbErrorWriteNonRetriable::IllegalState {
                            reason: "cannot append CancellationRequested event, execution is already cancelling".into(),
                            context: SpanTrace::capture(),
                            source: None,
                            loc: Location::caller(),
                        }
                        .into());
                    }
                    // Paused executions are 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::PendingAt(..)
                    | PendingState::BlockedByJoinSet(..)
                    | PendingState::Paused(..)
                    | PendingState::Locked(..) => {}
                }
                let next_version =
                    Self::update_state_cancelling(tx, execution_id, &appending_version)?;
                return Ok((next_version, AppendNotifier::default()));
            }

            ExecutionRequest::Finished { retval, .. } => {
                Self::update_state_finished(
                    tx,
                    execution_id,
                    &appending_version,
                    req.created_at,
                    PendingStateFinishedResultKind::from(retval),
                )?;
                return Ok((
                    appending_version,
                    AppendNotifier {
                        pending_at: None,
                        execution_finished: Some(NotifierExecutionFinished {
                            execution_id: execution_id.clone(),
                            retval: retval.clone(),
                        }),
                        response: None,
                    },
                ));
            }

            ExecutionRequest::HistoryEvent {
                event:
                    HistoryEvent::JoinSetCreate { .. }
                    | HistoryEvent::JoinSetRequest {
                        request: JoinSetRequest::ChildExecutionRequest { .. },
                        ..
                    }
                    | HistoryEvent::Persist { .. }
                    | HistoryEvent::Schedule { .. }
                    | HistoryEvent::Stub { .. }
                    | HistoryEvent::JoinNextTooMany { .. }
                    | HistoryEvent::JoinNextTry { .. },
            } => {
                return Ok((
                    Self::bump_state_next_version(tx, execution_id, &appending_version, None)?,
                    AppendNotifier::default(),
                ));
            }

            ExecutionRequest::HistoryEvent {
                event:
                    HistoryEvent::JoinSetRequest {
                        join_set_id,
                        request:
                            JoinSetRequest::DelayRequest {
                                delay_id,
                                expires_at,
                                paused,
                                ..
                            },
                    },
            } => {
                return Ok((
                    Self::bump_state_next_version(
                        tx,
                        execution_id,
                        &appending_version,
                        Some(DelayReq {
                            join_set_id: join_set_id.clone(),
                            delay_id: delay_id.clone(),
                            expires_at: *expires_at,
                            paused: *paused,
                        }),
                    )?,
                    AppendNotifier::default(),
                ));
            }

            ExecutionRequest::HistoryEvent {
                event:
                    HistoryEvent::JoinNext {
                        join_set_id,
                        run_expires_at,
                        closing,
                        requested_ffqn: _,
                    },
            } => {
                // Did the response arrive already?
                let join_next_count = Self::count_join_next(tx, execution_id, join_set_id)?;
                let nth_response =
                    Self::nth_response(tx, execution_id, join_set_id, join_next_count - 1)?; // Skip n-1 rows
                trace!("join_next_count: {join_next_count}, nth_response: {nth_response:?}");
                assert!(join_next_count > 0);
                if let Some(ResponseWithCursor {
                    event:
                        JoinSetResponseEventOuter {
                            created_at: nth_created_at,
                            ..
                        },
                    cursor: _,
                }) = nth_response
                {
                    let scheduled_at = max(*run_expires_at, nth_created_at); // No need to block
                    let (next_version, notifier) = Self::update_state_pending_after_event_appended(
                        tx,
                        execution_id,
                        &appending_version,
                        PendingAfterEventUpdate {
                            scheduled_at,
                            intermittent_failure: false,
                            component_input_digest: combined_state
                                .execution_with_state
                                .component_digest,
                        },
                    )?;
                    return Ok((next_version, notifier));
                }
                return Ok((
                    Self::update_state_blocked(
                        tx,
                        execution_id,
                        &appending_version,
                        join_set_id,
                        *run_expires_at,
                        *closing,
                    )?,
                    AppendNotifier::default(),
                ));
            }
        }
    }

    fn append_response(
        tx: &Transaction,
        execution_id: &ExecutionId,
        event: JoinSetResponseEventOuter,
    ) -> Result<AppendNotifier, DbErrorWrite> {
        // SQLite has a single global writer, so the read-modify-write of MAX(seq)+1
        // is serialized without extra locking; `seq` is contiguous and 1-based per execution.
        let mut stmt = tx.prepare(
            "INSERT INTO t_join_set_response (execution_id, created_at, join_set_id, delay_id, delay_success, child_execution_id, finished_version, seq) \
                    VALUES (:execution_id, :created_at, :join_set_id, :delay_id, :delay_success, :child_execution_id, :finished_version, \
                    (SELECT COALESCE(MAX(seq), 0) + 1 FROM t_join_set_response WHERE execution_id = :execution_id)) \
                    RETURNING seq",
        )?;
        let join_set_id = &event.event.join_set_id;
        let (delay_id, delay_success) = match &event.event.event {
            JoinSetResponse::DelayFinished { delay_id, result } => {
                (Some(delay_id.to_string()), Some(result.is_ok()))
            }
            JoinSetResponse::ChildExecutionFinished { .. } => (None, None),
        };
        let (child_execution_id, finished_version) = match &event.event.event {
            JoinSetResponse::ChildExecutionFinished {
                child_execution_id,
                finished_version,
                result: _,
            } => (
                Some(child_execution_id.to_string()),
                Some(finished_version.0),
            ),
            JoinSetResponse::DelayFinished { .. } => (None, None),
        };

        let seq = stmt.query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
                ":created_at": event.created_at,
                ":join_set_id": join_set_id.to_string(),
                ":delay_id": delay_id,
                ":delay_success": delay_success,
                ":child_execution_id": child_execution_id,
                ":finished_version": finished_version,
            },
            |row| row.get::<_, u32>(0),
        )?;
        let cursor = ResponseCursor(seq);

        // if the execution is going to be unblocked by this response...
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        debug!("previous_pending_state: {combined_state:?}");
        // A cancelling execution is finished by the cancellation driver; it is never
        // picked up by executors again, so do not unblock or notify it.
        let mut notifier = if let PendingStateMerged::BlockedByJoinSet {
            state:
                PendingStateBlockedByJoinSet {
                    join_set_id: found_join_set_id,
                    lock_expires_at, // Set to a future time if the worker is keeping the execution warm waiting for the result.
                    closing: _,
                },
            lifecycle: Lifecycle::Active | Lifecycle::Paused,
        } =
            PendingStateMerged::from(combined_state.execution_with_state.pending_state)
            && *join_set_id == found_join_set_id
        {
            // PendingAt should be set to current time if called from expired_timers_watcher,
            // or to a future time if the execution is hot.
            let scheduled_at = max(lock_expires_at, event.created_at);
            // TODO: Add diff test
            // Unblock the state.
            Self::update_state_pending_after_response_appended(
                tx,
                execution_id,
                scheduled_at,
                combined_state.execution_with_state.component_digest,
            )?
        } else {
            AppendNotifier::default()
        };
        if let JoinSetResponseEvent {
            join_set_id,
            event:
                JoinSetResponse::DelayFinished {
                    delay_id,
                    result: _,
                },
        } = &event.event
        {
            debug!(%join_set_id, %delay_id, "Deleting from `t_delay`");
            let mut stmt =
                tx.prepare_cached("DELETE FROM t_delay WHERE execution_id = :execution_id AND join_set_id = :join_set_id AND delay_id = :delay_id")
                ?;
            stmt.execute(named_params! {
                ":execution_id": execution_id.to_string(),
                ":join_set_id": join_set_id.to_string(),
                ":delay_id": delay_id.to_string(),
            })?;
        }
        notifier.response = Some((execution_id.clone(), ResponseWithCursor { cursor, event }));
        Ok(notifier)
    }

    fn append_backtrace(
        tx: &Transaction,
        backtrace_info: &BacktraceInfo,
    ) -> Result<usize, DbErrorWrite> {
        let backtrace_hash = backtrace_info.wasm_backtrace.hash();

        tx.prepare("INSERT OR IGNORE INTO t_wasm_backtrace (backtrace_hash, wasm_backtrace) VALUES (:backtrace_hash, :wasm_backtrace)")?
        .execute(named_params! {
            ":backtrace_hash": backtrace_hash,
            ":wasm_backtrace": JsonWrapper(&backtrace_info.wasm_backtrace)
        })?;

        tx.prepare(
                "INSERT OR IGNORE INTO t_execution_backtrace (execution_id, component_id, version_min_including, version_max_excluding, backtrace_hash) \
                    VALUES (:execution_id, :component_id, :version_min_including, :version_max_excluding, :backtrace_hash)",
        )?
        .execute(named_params! {
            ":execution_id": backtrace_info.execution_id.to_string(),
            ":component_id": JsonWrapper(&backtrace_info.component_id),
            ":version_min_including": backtrace_info.version_min_including.0,
            ":version_max_excluding": backtrace_info.version_max_excluding.0,
            ":backtrace_hash": backtrace_hash,
        })
        .map_err(DbErrorWrite::from)
    }

    fn append_log(tx: &Transaction, row: &LogInfoAppendRow) -> Result<(), DbErrorWrite> {
        let mut stmt = tx.prepare(
            "INSERT INTO t_log (
            execution_id,
            run_id,
            created_at,
            level,
            message,
            stream_type,
            payload
        ) VALUES (
            :execution_id,
            :run_id,
            :created_at,
            :level,
            :message,
            :stream_type,
            :payload
        )",
        )?;

        match &row.log_entry {
            LogEntry::Log {
                created_at,
                level,
                message,
            } => {
                stmt.execute(named_params! {
                    ":execution_id": row.execution_id,
                    ":run_id": row.run_id,
                    ":created_at": created_at,
                    ":level": *level as u8,
                    ":message": message,
                    ":stream_type": Option::<u8>::None,
                    ":payload": Option::<Vec<u8>>::None,
                })?;
            }
            LogEntry::Stream {
                created_at,
                payload,
                stream_type,
            } => {
                stmt.execute(named_params! {
                    ":execution_id": row.execution_id,
                    ":run_id": row.run_id,
                    ":created_at": created_at,
                    ":level": Option::<u8>::None,
                    ":message": Option::<String>::None,
                    ":stream_type": *stream_type as u8,
                    ":payload": payload,
                })?;
            }
        }

        Ok(())
    }

    fn get(
        tx: &Transaction,
        execution_id: &ExecutionId,
    ) -> Result<concepts::storage::ExecutionLog, DbErrorRead> {
        let mut stmt = tx.prepare(
            "SELECT created_at, json_value, version FROM t_execution_log WHERE \
                        execution_id = :execution_id ORDER BY version",
        )?;
        let events = stmt
            .query_map(
                named_params! {
                    ":execution_id": execution_id.to_string(),
                },
                |row| {
                    let created_at = row.get("created_at")?;
                    let event = row
                        .get::<_, JsonWrapper<ExecutionRequest>>("json_value")
                        .map_err(|serde| {
                            error!("Cannot deserialize {row:?} - {serde:?}");
                            consistency_rusqlite("cannot deserialize event")
                        })?
                        .0;
                    let version = Version(row.get("version")?);

                    Ok(ExecutionEvent {
                        created_at,
                        event,
                        backtrace_id: None,
                        version,
                    })
                },
            )?
            .collect::<Result<Vec<_>, _>>()?;
        if events.is_empty() {
            return Err(DbErrorRead::NotFound);
        }
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        let responses = Self::list_responses(tx, execution_id, None)?;
        Ok(concepts::storage::ExecutionLog {
            execution_id: execution_id.clone(),
            events,
            responses,
            next_version: combined_state.get_next_version_or_finished(), // In case of finished, this will be the already last version
            pending_state: combined_state.execution_with_state.pending_state,
            component_digest: combined_state.execution_with_state.component_digest,
            component_type: combined_state.execution_with_state.component_type,
            deployment_id: combined_state.execution_with_state.deployment_id,
        })
    }

    fn get_max_version(
        tx: &Transaction,
        execution_id: &ExecutionId,
    ) -> Result<Version, DbErrorRead> {
        tx.prepare("SELECT MAX(version) FROM t_execution_log WHERE execution_id = :execution_id")?
            .query_row(
                named_params! { ":execution_id": execution_id.to_string() },
                |row| row.get::<_, Option<VersionType>>(0),
            )
            .map(|v| v.map(Version::new).ok_or(DbErrorRead::NotFound))
            .map_err(DbErrorRead::from)
            .flatten()
    }

    fn get_max_response_cursor(
        tx: &Transaction,
        execution_id: &ExecutionId,
    ) -> Result<ResponseCursor, DbErrorRead> {
        let max_cursor = tx
            .prepare("SELECT MAX(seq) FROM t_join_set_response WHERE execution_id = :execution_id")?
            .query_row(
                named_params! { ":execution_id": execution_id.to_string() },
                |row| row.get::<_, Option<u32>>(0),
            )?;
        // Assume the execution exists and has no responses
        let max_cursor = max_cursor.unwrap_or_default();
        Ok(ResponseCursor(max_cursor))
    }

    fn list_execution_events(
        tx: &Transaction,
        execution_id: &ExecutionId,
        pagination: Pagination<VersionType>,
        include_backtrace_id: bool,
    ) -> Result<Vec<ExecutionEvent>, DbErrorRead> {
        let mut params: Vec<(&'static str, Box<dyn rusqlite::ToSql>)> = vec![];
        params.push((":execution_id", Box::new(execution_id.to_string())));

        let (cursor, length, rel, is_desc) = match &pagination {
            Pagination::NewerThan {
                cursor,
                length,
                including_cursor,
            } => (
                *cursor,
                *length,
                if *including_cursor { ">=" } else { ">" },
                false,
            ),
            Pagination::OlderThan {
                cursor,
                length,
                including_cursor,
            } => (
                *cursor,
                *length,
                if *including_cursor { "<=" } else { "<" },
                true,
            ),
        };
        params.push((":cursor", Box::new(cursor)));

        let base_select = if include_backtrace_id {
            format!(
                "SELECT
                    log.created_at,
                    log.json_value,
                    log.version as version,
                    bt.version_min_including AS backtrace_id
                FROM
                    t_execution_log AS log
                LEFT OUTER JOIN
                    t_execution_backtrace AS bt ON log.execution_id = bt.execution_id
                                    AND log.version >= bt.version_min_including
                                    AND log.version < bt.version_max_excluding
                WHERE
                    log.execution_id = :execution_id
                    AND log.version {rel} :cursor"
            )
        } else {
            format!(
                "SELECT
                    created_at, json_value, NULL as backtrace_id, version
                FROM t_execution_log WHERE
                    execution_id = :execution_id AND version {rel} :cursor"
            )
        };

        let order = if is_desc { "DESC" } else { "ASC" };
        let mut sql = format!("{base_select} ORDER BY version {order} LIMIT {length}");

        // Re-order to ascending for consistent oldest-to-newest results
        if is_desc {
            sql = format!("SELECT * FROM ({sql}) ORDER BY version ASC");
        }

        tx.prepare(&sql)?
            .query_map::<_, &[(&'static str, &dyn ToSql)], _>(
                params
                    .iter()
                    .map(|(key, value)| (*key, value.as_ref()))
                    .collect::<Vec<_>>()
                    .as_ref(),
                |row| {
                    let created_at = row.get("created_at")?;
                    let backtrace_id = row
                        .get::<_, Option<VersionType>>("backtrace_id")?
                        .map(Version::new);
                    let version = Version(row.get("version")?);

                    let event = row
                        .get::<_, JsonWrapper<ExecutionRequest>>("json_value")
                        .map(|event| ExecutionEvent {
                            created_at,
                            event: event.0,
                            backtrace_id,
                            version,
                        })
                        .map_err(|serde| {
                            error!("Cannot deserialize {row:?} - {serde:?}");
                            consistency_rusqlite("cannot deserialize")
                        })?;
                    Ok(event)
                },
            )?
            .collect::<Result<Vec<_>, _>>()
            .map_err(DbErrorRead::from)
    }

    fn map_t_execution_log_row(row: &Row<'_>) -> Result<ExecutionEvent, rusqlite::Error> {
        let created_at = row.get("created_at")?;
        let event = row
            .get::<_, JsonWrapper<ExecutionRequest>>("json_value")
            .map_err(|serde| {
                error!("Cannot deserialize {row:?} - {serde:?}");
                consistency_rusqlite("cannot deserialize event")
            })?;
        let version = Version(row.get("version")?);

        Ok(ExecutionEvent {
            created_at,
            event: event.0,
            backtrace_id: None,
            version,
        })
    }

    fn get_execution_event(
        tx: &Transaction,
        execution_id: &ExecutionId,
        version: VersionType,
    ) -> Result<ExecutionEvent, DbErrorRead> {
        tx.prepare(
            "SELECT created_at, json_value, version FROM t_execution_log WHERE \
                        execution_id = :execution_id AND version = :version",
        )?
        .query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
                ":version": version,
            },
            SqlitePool::map_t_execution_log_row,
        )
        .map_err(DbErrorRead::from)
    }

    fn get_last_execution_event(
        tx: &Transaction,
        execution_id: &ExecutionId,
    ) -> Result<ExecutionEvent, DbErrorRead> {
        tx.prepare(
            "SELECT created_at, json_value, version FROM t_execution_log WHERE \
                        execution_id = :execution_id ORDER BY version DESC LIMIT 1",
        )?
        .query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
            },
            SqlitePool::map_t_execution_log_row,
        )
        .map_err(DbErrorRead::from)
    }

    fn get_delay_response(
        tx: &Transaction,
        execution_id: &ExecutionId,
        delay_id: &DelayId,
    ) -> Result<Option<bool>, DbErrorRead> {
        // TODO: Add test
        tx.prepare(
            "SELECT delay_success \
                    FROM t_join_set_response \
                    WHERE \
                    execution_id = :execution_id AND delay_id = :delay_id
                    ",
        )?
        .query_row(
            named_params! {
                ":execution_id": execution_id.to_string(),
                ":delay_id": delay_id.to_string(),
            },
            |row| {
                let delay_success = row.get::<_, bool>("delay_success")?;
                Ok(delay_success)
            },
        )
        .optional()
        .map_err(DbErrorRead::from)
    }

    #[instrument(level = Level::TRACE, skip_all)]
    /// Find next responses for this execution
    fn get_responses_after(
        tx: &Transaction,
        execution_id: &ExecutionId,
        last_response: ResponseCursor,
    ) -> Result<Vec<ResponseWithCursor>, DbErrorRead> {
        // TODO: Add test
        tx.prepare(
            "SELECT r.id, r.seq, r.created_at, r.join_set_id, \
            r.delay_id, r.delay_success, \
            r.child_execution_id, r.finished_version, child.json_value \
            FROM t_join_set_response r LEFT OUTER JOIN t_execution_log child ON r.child_execution_id = child.execution_id \
            WHERE \
            r.seq > :last_response_seq AND \
            r.execution_id = :execution_id AND \
            ( \
            r.finished_version = child.version \
            OR r.child_execution_id IS NULL \
            ) \
            ORDER BY seq",
        )
        ?
        .query_map(
            named_params! {
                ":last_response_seq": last_response.0,
                ":execution_id": execution_id.to_string(),
            },
            Self::parse_response_with_cursor,
        )
        ?
        .collect::<Result<Vec<_>, _>>()
        .map_err(DbErrorRead::from)
    }

    fn get_pending_of_single_ffqn(
        mut stmt: CachedStatement,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        ffqn: &FunctionFqn,
    ) -> Result<Vec<(ExecutionId, Version)>, ()> {
        stmt.query_map(
            named_params! {
                ":pending_expires_finished": pending_at_or_sooner,
                ":ffqn": ffqn.to_string(),
                ":batch_size": batch_size,
            },
            |row| {
                let execution_id = row.get::<_, ExecutionId>("execution_id")?;
                let next_version =
                    Version::new(row.get::<_, VersionType>("corresponding_version")?).increment();
                Ok((execution_id, next_version))
            },
        )
        .map_err(|err| {
            warn!("Ignoring consistency error {err:?}");
        })?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|err| {
            warn!("Ignoring consistency error {err:?}");
        })
    }

    /// Get executions and their next versions
    fn get_pending_by_ffqns(
        conn: &Connection,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        ffqns: &[FunctionFqn],
    ) -> Result<Vec<(ExecutionId, Version)>, RusqliteError> {
        let batch_size = usize::try_from(batch_size).expect("16 bit systems are unsupported");
        let mut execution_ids_versions = Vec::with_capacity(batch_size);
        for ffqn in ffqns {
            let needed = batch_size - execution_ids_versions.len();
            if needed == 0 {
                break;
            }
            let needed =
                u32::try_from(needed).expect("`batch_size`:u32 - usize cannot overflow an u32");
            // Select executions in PendingAt.
            let stmt = conn.prepare_cached(&format!(
                r#"
                    SELECT execution_id, corresponding_version FROM t_state WHERE
                    state = "{STATE_PENDING_AT}" AND
                    pending_expires_finished <= :pending_expires_finished AND ffqn = :ffqn
                    AND lifecycle = 'active'
                    ORDER BY pending_expires_finished LIMIT :batch_size
                    "#
            ))?;

            if let Ok(execs_and_versions) =
                Self::get_pending_of_single_ffqn(stmt, needed, pending_at_or_sooner, ffqn)
            {
                execution_ids_versions.extend(execs_and_versions);
                // consistency errors are ignored since we want to return at least some rows.
            }
        }
        Ok(execution_ids_versions)
    }

    fn get_pending_by_ffqns_auto(
        conn: &Connection,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        ffqns: &[FunctionFqn],
        current_digest: &ComponentDigest,
    ) -> Result<Vec<(ExecutionId, Version)>, RusqliteError> {
        let batch_size = usize::try_from(batch_size).expect("16 bit systems are unsupported");
        let mut execution_ids_versions = Vec::with_capacity(batch_size);
        for ffqn in ffqns {
            let needed = batch_size - execution_ids_versions.len();
            if needed == 0 {
                break;
            }
            let mut stmt = conn.prepare_cached(&format!(
                r#"
                    SELECT execution_id, corresponding_version FROM t_state WHERE
                    state = "{STATE_PENDING_AT}" AND
                    pending_expires_finished <= :pending_expires_finished AND ffqn = :ffqn
                    AND lifecycle = 'active'
                    AND (incompatible_digest IS NULL OR incompatible_digest <> :current_digest)
                    ORDER BY pending_expires_finished LIMIT :batch_size
                    "#
            ))?;

            if let Ok(execs_and_versions) = stmt
                .query_map(
                    named_params! {
                        ":pending_expires_finished": pending_at_or_sooner,
                        ":ffqn": ffqn.to_string(),
                        ":current_digest": current_digest,
                        ":batch_size": u32::try_from(needed)
                            .expect("`needed` is <= `batch_size` which is u32"),
                    },
                    |row| {
                        let execution_id = row.get::<_, ExecutionId>("execution_id")?;
                        let next_version =
                            Version::new(row.get::<_, VersionType>("corresponding_version")?)
                                .increment();
                        Ok((execution_id, next_version))
                    },
                )
                .and_then(|rows| rows.collect::<Result<Vec<_>, _>>())
            {
                execution_ids_versions.extend(execs_and_versions);
                if execution_ids_versions.len() == batch_size {
                    break;
                }
            }
        }
        Ok(execution_ids_versions)
    }

    fn get_pending_by_component_input_digest(
        conn: &Connection,
        batch_size: u32,
        pending_at_or_sooner: DateTime<Utc>,
        input_digest: &ComponentDigest,
    ) -> Result<Vec<(ExecutionId, Version)>, RusqliteError> {
        let mut stmt = conn.prepare_cached(&format!(
            r#"
                SELECT execution_id, corresponding_version FROM t_state WHERE
                state = "{STATE_PENDING_AT}" AND
                pending_expires_finished <= :pending_expires_finished AND
                component_id_input_digest = :component_id_input_digest
                AND lifecycle = 'active'
                ORDER BY pending_expires_finished LIMIT :batch_size
                "#
        ))?;

        stmt.query_map(
            named_params! {
                ":pending_expires_finished": pending_at_or_sooner,
                ":component_id_input_digest": input_digest,
                ":batch_size": batch_size,
            },
            |row| {
                let execution_id = row.get::<_, ExecutionId>("execution_id")?;
                let next_version =
                    Version::new(row.get::<_, VersionType>("corresponding_version")?).increment();
                Ok((execution_id, next_version))
            },
        )?
        .collect::<Result<Vec<_>, _>>()
        .map_err(RusqliteError::from)
    }

    // Must be called after write transaction for a correct happens-before relationship.
    #[instrument(level = Level::TRACE, skip_all)]
    fn notify_all(&self, notifiers: Vec<AppendNotifier>, current_time: DateTime<Utc>) {
        let (pending_ats, finished_execs, responses) = {
            let (mut pending_ats, mut finished_execs, mut responses) =
                (Vec::new(), Vec::new(), Vec::new());
            for notifier in notifiers {
                if let Some(pending_at) = notifier.pending_at {
                    pending_ats.push(pending_at);
                }
                if let Some(finished) = notifier.execution_finished {
                    finished_execs.push(finished);
                }
                if let Some(response) = notifier.response {
                    responses.push(response);
                }
            }
            (pending_ats, finished_execs, responses)
        };

        // Notify pending_at subscribers.
        if !pending_ats.is_empty() {
            let guard = self.0.pending_subscribers.lock().unwrap();
            for pending_at in pending_ats {
                Self::notify_pending_locked(&pending_at, current_time, &guard);
            }
        }
        // Notify execution finished subscribers.
        // Every NotifierExecutionFinished value belongs to a different execution, since only `append(Finished)` can produce `NotifierExecutionFinished`.
        if !finished_execs.is_empty() {
            let mut guard = self.0.execution_finished_subscribers.lock().unwrap();
            for finished in finished_execs {
                if let Some(listeners_of_exe_id) = guard.remove(&finished.execution_id) {
                    for (_tag, sender) in listeners_of_exe_id {
                        // Sending while holding the lock but the oneshot sender does not block.
                        // If `wait_for_finished_result` happens after the append, it would receive the finished value instead.
                        let _ = sender.send(finished.retval.clone());
                    }
                }
            }
        }
        // Notify response subscribers.
        if !responses.is_empty() {
            let mut guard = self.0.response_subscribers.lock().unwrap();
            for (execution_id, _response) in responses {
                if let Some((sender, _)) = guard.remove(&execution_id) {
                    let _ = sender.send(());
                }
            }
        }
    }

    fn notify_pending_locked(
        notifier: &NotifierPendingAt,
        current_time: DateTime<Utc>,
        ffqn_to_pending_subscription: &std::sync::MutexGuard<PendingFfqnSubscribersHolder>,
    ) {
        // No need to remove here, cleanup is handled by the caller.
        if notifier.scheduled_at <= current_time {
            ffqn_to_pending_subscription.notify(notifier);
        }
    }

    fn upgrade_execution_component_single_write(
        tx: &Transaction,
        execution_id: &ExecutionId,
        old: &ComponentDigest,
        new: &ComponentDigest,
        reason: ComponentUpgradeReason,
    ) -> Result<(), DbErrorWrite> {
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        if combined_state.execution_with_state.component_digest != *old {
            return Err(DbErrorWrite::NotFound);
        }
        let appending_version = combined_state.get_next_version_fail_if_finished()?;
        Self::append(
            tx,
            execution_id,
            AppendRequest {
                created_at: Utc::now(),
                event: ExecutionRequest::ComponentUpgradeFinished {
                    component_digest: new.clone(),
                    deployment_id: combined_state.execution_with_state.deployment_id,
                    outcome: ComponentUpgradeOutcome::Success { reason },
                },
            },
            appending_version,
        )?;
        Ok(())
    }

    fn list_logs_tx(
        tx: &Transaction,
        execution_id: &ExecutionId,
        show_derived: bool,
        filter: &LogFilter,
        pagination: &Pagination<LogCursor>,
    ) -> Result<ListLogsResponse, DbErrorRead> {
        let length = pagination.length();
        let exec_id_str = execution_id.to_string();
        let exec_id_filter = if show_derived {
            "LIKE :execution_id || '%'"
        } else {
            "= :execution_id"
        };
        let mut query = format!(
            "SELECT id, run_id, created_at, level, message, stream_type, payload, execution_id
             FROM t_log
             WHERE execution_id {exec_id_filter}",
        );

        let cursor = pagination.cursor();
        let created_after = filter.created_after();
        let created_before = filter.created_before();
        let mut params = vec![
            (":execution_id", &exec_id_str as &dyn rusqlite::ToSql),
            (":cursor", &cursor.0 as &dyn rusqlite::ToSql),
            (":length", &length as &dyn rusqlite::ToSql),
        ];
        if let Some(created_after) = &created_after {
            params.push((":created_after", created_after as &dyn rusqlite::ToSql));
        }
        if let Some(created_before) = &created_before {
            params.push((":created_before", created_before as &dyn rusqlite::ToSql));
        }

        // Logs and streams filter
        let level_filter = if filter.should_show_logs() {
            let levels_str = if !filter.levels().is_empty() {
                filter
                    .levels()
                    .iter()
                    .map(|lvl| (*lvl as u8).to_string())
                    .collect::<Vec<_>>()
                    .join(",")
            } else {
                LogLevel::iter()
                    .map(|lvl| (lvl as u8).to_string())
                    .collect::<Vec<_>>()
                    .join(",")
            };
            Some(format!(" level IN ({levels_str})"))
        } else {
            None
        };
        let stream_filter = if filter.should_show_streams() {
            let streams_str = if !filter.stream_types().is_empty() {
                filter
                    .stream_types()
                    .iter()
                    .map(|st| (*st as u8).to_string())
                    .collect::<Vec<_>>()
                    .join(",")
            } else {
                LogStreamType::iter()
                    .map(|st| (st as u8).to_string())
                    .collect::<Vec<_>>()
                    .join(",")
            };
            Some(format!(" stream_type IN ({streams_str})"))
        } else {
            None
        };
        match (level_filter, stream_filter) {
            (Some(level_filter), Some(stream_filter)) => {
                write!(&mut query, " AND ({level_filter} OR {stream_filter})")
                    .expect("writing to string");
            }
            (Some(level_filter), None) => {
                write!(&mut query, " AND {level_filter}").expect("writing to string");
            }
            (None, Some(stream_filter)) => {
                write!(&mut query, " AND {stream_filter}").expect("writing to string");
            }
            (None, None) => unreachable!("guarded by constructor"),
        }

        if created_after.is_some() {
            query.push_str(" AND created_at > :created_after");
        }
        if created_before.is_some() {
            query.push_str(" AND created_at < :created_before");
        }

        // Pagination and ordering
        write!(&mut query, " AND id {} :cursor", pagination.rel()).expect("writing to string");
        query.push_str(" ORDER BY id ");
        query.push_str(pagination.asc_or_desc());

        // Limit
        query.push_str(" LIMIT :length");

        let mut stmt = tx.prepare(&query)?;

        let items = stmt
            .query_map(params.as_slice(), |row| {
                let created_at: DateTime<Utc> = row.get("created_at")?;
                let run_id = row.get("run_id")?;
                let level: Option<u8> = row.get("level")?;
                let message: Option<String> = row.get("message")?;
                let stream_type: Option<u8> = row.get("stream_type")?;
                let payload: Option<Vec<u8>> = row.get("payload")?;
                let execution_id_str: String = row.get("execution_id")?;
                let execution_id = ExecutionId::from_str(&execution_id_str).map_err(|_| {
                    consistency_rusqlite(format!("cannot convert ExecutionId {execution_id_str}"))
                })?;

                let log_entry = match (level, message, stream_type, payload) {
                    (Some(lvl), Some(msg), None, None) => LogEntry::Log {
                        created_at,
                        level: LogLevel::try_from(lvl).map_err(|_| {
                            consistency_rusqlite(format!("cannot convert {lvl} to LogLevel"))
                        })?,
                        message: msg,
                    },
                    (None, None, Some(stype), Some(pl)) => LogEntry::Stream {
                        created_at,
                        stream_type: LogStreamType::try_from(stype).map_err(|_| {
                            consistency_rusqlite(format!("cannot convert {stype} to LogStreamType"))
                        })?,
                        payload: pl,
                    },
                    _ => {
                        return Err(consistency_rusqlite("invalid t_log row".to_string()));
                    }
                };
                Ok(LogEntryRow {
                    cursor: LogCursor(row.get("id")?),
                    run_id,
                    log_entry,
                    execution_id,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;

        Ok(ListLogsResponse {
            next_page: items
                .last()
                .map(|item| Pagination::NewerThan {
                    length: pagination.length(),
                    cursor: item.cursor,
                    including_cursor: false,
                })
                .unwrap_or({
                    if pagination.is_asc() {
                        *pagination // no new results, keep the same cursor
                    } else {
                        // no prev results, let's start from beginning
                        Pagination::NewerThan {
                            length: pagination.length(),
                            cursor: LogCursor(i64::MIN),
                            including_cursor: false,
                        }
                    }
                }),
            prev_page: match items.first() {
                Some(item) => Some(Pagination::OlderThan {
                    length: pagination.length(),
                    cursor: item.cursor,
                    including_cursor: false,
                }),
                None if pagination.is_asc() && pagination.cursor() != &LogCursor(i64::MIN) => {
                    // asked for a next page that does not exists (yet).
                    Some(pagination.invert())
                }
                None => None,
            },
            items,
        })
    }

    fn list_deployment_states(
        tx: &Transaction,
        current_time: DateTime<Utc>,
        pagination: Pagination<Option<DeploymentId>>,
        include_deployment_toml: bool,
        execution_counts: DeploymentExecutionCounts,
    ) -> Result<Vec<DeploymentState>, DbErrorRead> {
        let mut params: Vec<(&'static str, Box<dyn ToSql>)> = vec![];
        let deployment_toml_col = if include_deployment_toml {
            "d.deployment_toml"
        } else {
            "NULL AS deployment_toml"
        };
        let include_execution_counts =
            matches!(execution_counts, DeploymentExecutionCounts::Count { .. });
        // Without counts, skip the SUMs (and the `t_state` join feeding them) and report zeros.
        let count_cols = if include_execution_counts {
            format!(
                r"
            COALESCE(SUM(s.state = '{STATE_LOCKED}' AND s.lifecycle = 'active'), 0) AS locked,
            COALESCE(SUM(s.state = '{STATE_PENDING_AT}' AND s.lifecycle = 'active' AND s.pending_expires_finished <= :now), 0) AS pending,
            COALESCE(SUM(s.state = '{STATE_PENDING_AT}' AND s.lifecycle = 'active' AND s.pending_expires_finished > :now), 0) AS scheduled,
            COALESCE(SUM(s.state = '{STATE_BLOCKED_BY_JOIN_SET}' AND s.lifecycle = 'active'), 0) AS blocked,
            COALESCE(SUM(s.lifecycle = 'paused'), 0) AS paused,
            COALESCE(SUM(s.lifecycle = 'cancelling'), 0) AS cancelling,
            COALESCE(SUM(s.state = '{STATE_FINISHED}' AND s.result_kind = '{RESULT_KIND_JSON_OK}'), 0) AS finished_ok,
            COALESCE(SUM(s.state = '{STATE_FINISHED}' AND s.result_kind = '{RESULT_KIND_JSON_ERROR}'), 0) AS finished_error,
            COALESCE(SUM(s.state = '{STATE_FINISHED}' AND s.result_kind IS NOT NULL
                AND s.result_kind NOT IN ('{RESULT_KIND_JSON_OK}', '{RESULT_KIND_JSON_ERROR}')), 0) AS finished_execution_failure,"
            )
        } else {
            "
            0 AS locked,
            0 AS pending,
            0 AS scheduled,
            0 AS blocked,
            0 AS paused,
            0 AS cancelling,
            0 AS finished_ok,
            0 AS finished_error,
            0 AS finished_execution_failure,"
                .to_string()
        };
        let mut sql = format!(
            r"
        SELECT
            d.deployment_id,
            d.description,
            d.digest,{count_cols}
            {deployment_toml_col},
            d.created_at,
            d.last_active_at,
            d.status
        FROM t_deployment d{join}",
            join = match execution_counts {
                DeploymentExecutionCounts::Count { include_derived } => {
                    let join_top_level = if include_derived {
                        ""
                    } else {
                        " AND s.is_top_level = true"
                    };
                    format!(
                        "\n        LEFT JOIN t_state s ON s.deployment_id = d.deployment_id{join_top_level}"
                    )
                }
                DeploymentExecutionCounts::Skip => String::new(),
            }
        );

        if include_execution_counts {
            params.push((":now", Box::new(current_time)));
        }

        if let Some(cursor) = pagination.cursor() {
            params.push((":cursor", Box::new(*cursor)));
            write!(
                sql,
                " WHERE d.deployment_id {rel} :cursor",
                rel = pagination.rel()
            )
            .expect("writing to string");
        }

        // Inner query: fetch rows with cursor-based ordering
        // Outer query: always return results in descending order
        let (inner_order, outer_order) = if pagination.is_desc() {
            ("DESC", "")
        } else {
            ("ASC", "DESC")
        };

        // GROUP BY only needed to collapse the rows multiplied by the `t_state` join.
        if include_execution_counts {
            write!(
                sql,
                " GROUP BY d.deployment_id, d.description, d.digest, d.deployment_toml, d.created_at, d.last_active_at, d.status"
            )
            .expect("writing to string");
        }
        write!(
            sql,
            " ORDER BY d.deployment_id {inner_order} LIMIT {limit}",
            limit = pagination.length()
        )
        .expect("writing to string");

        let final_sql = if outer_order.is_empty() {
            sql
        } else {
            format!("SELECT * FROM ({sql}) AS sub ORDER BY deployment_id {outer_order}")
        };

        let result: Vec<DeploymentState> = tx
            .prepare(&final_sql)?
            .query_map::<_, &[(&'static str, &dyn ToSql)], _>(
                params
                    .iter()
                    .map(|(k, v)| (*k, v.as_ref()))
                    .collect::<Vec<_>>()
                    .as_ref(),
                |row| {
                    let status_str: String = row.get("status")?;
                    let status = status_str.parse::<DeploymentStatus>().map_err(|_| {
                        rusqlite::Error::InvalidColumnType(
                            0,
                            "status".to_string(),
                            rusqlite::types::Type::Text,
                        )
                    })?;
                    Ok(DeploymentState {
                        deployment_id: row.get("deployment_id")?,
                        description: row.get("description")?,
                        digest: row.get("digest")?,
                        locked: row.get("locked")?,
                        pending: row.get("pending")?,
                        scheduled: row.get("scheduled")?,
                        blocked: row.get("blocked")?,
                        paused: row.get("paused")?,
                        cancelling: row.get("cancelling")?,
                        finished_ok: row.get("finished_ok")?,
                        finished_error: row.get("finished_error")?,
                        finished_execution_failure: row.get("finished_execution_failure")?,
                        deployment_toml: row.get("deployment_toml")?,
                        created_at: row.get("created_at")?,
                        last_active_at: row.get("last_active_at")?,
                        status,
                    })
                },
            )?
            .collect::<Result<Vec<_>, rusqlite::Error>>()
            .map_err(DbErrorRead::from)?;

        Ok(result)
    }

    fn insert_deployment_tx(
        tx: &Transaction,
        record: &DeploymentRecord,
    ) -> Result<(), DbErrorWrite> {
        assert_eq!(
            record.status,
            DeploymentStatus::Inactive,
            "insert_deployment requires Inactive status"
        );
        assert!(
            record.last_active_at.is_none(),
            "insert_deployment requires last_active_at == None"
        );
        tx.execute(
            "INSERT INTO t_deployment \
             (deployment_id, description, digest, created_at, status, deployment_toml, obelisk_version, created_by) \
             VALUES (:deployment_id, :description, :digest, :created_at, :status, :deployment_toml, :obelisk_version, :created_by)",
            rusqlite::named_params! {
                ":deployment_id": record.deployment_id.to_string(),
                ":description": record.description,
                ":digest": record.digest.to_string(),
                ":created_at": record.created_at,
                ":status": record.status.as_str(),
                ":deployment_toml": record.deployment_toml,
                ":obelisk_version": record.obelisk_version,
                ":created_by": record.created_by,
            },
        )
        .map_err(RusqliteError::from)?;
        Self::insert_deployment_files_tx(tx, record.deployment_id, &record.files)?;
        Ok(())
    }

    fn insert_deployment_files_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
        files: &[DeploymentFileRecord],
    ) -> Result<(), DbErrorWrite> {
        let mut stmt = tx
            .prepare_cached(
                "INSERT INTO t_deployment_file (deployment_id, digest, path) \
                 VALUES (:deployment_id, :digest, :path)",
            )
            .map_err(RusqliteError::from)?;
        for file in files {
            stmt.execute(rusqlite::named_params! {
                ":deployment_id": deployment_id.to_string(),
                ":digest": file.digest.to_string(),
                ":path": file.path,
            })
            .map_err(RusqliteError::from)?;
        }
        Ok(())
    }

    fn upsert_component_metadata_tx(
        tx: &Transaction,
        records: &[ComponentMetadataRecord],
    ) -> Result<(), DbErrorWrite> {
        let mut stmt = tx
            .prepare(
                "INSERT OR IGNORE INTO t_component_metadata \
                 (component_digest, imports_json, exports_json, wit, wit_origin) \
                 VALUES (:component_digest, :imports_json, :exports_json, :wit, :wit_origin)",
            )
            .map_err(RusqliteError::from)?;
        for record in records {
            let imports_json = serde_json::to_string(&record.imports).map_err(|err| {
                RusqliteError::from(rusqlite::Error::ToSqlConversionFailure(Box::new(err)))
            })?;
            let exports_json = serde_json::to_string(&record.exports).map_err(|err| {
                RusqliteError::from(rusqlite::Error::ToSqlConversionFailure(Box::new(err)))
            })?;
            stmt.execute(named_params! {
                ":component_digest": record.component_digest.clone(),
                ":imports_json": imports_json,
                ":exports_json": exports_json,
                ":wit": record.wit.clone(),
                ":wit_origin": record.wit_origin.clone(),
            })
            .map_err(RusqliteError::from)?;
        }
        Ok(())
    }

    fn insert_deployment_components_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
        records: &[DeploymentComponentRecord],
    ) -> Result<(), DbErrorWrite> {
        let mut stmt = tx
            .prepare(
                "INSERT OR IGNORE INTO t_deployment_component \
                 (deployment_id, component_name, component_type, component_digest) \
                 VALUES (:deployment_id, :component_name, :component_type, :component_digest)",
            )
            .map_err(RusqliteError::from)?;
        for record in records {
            debug_assert_eq!(record.deployment_id, deployment_id);
            stmt.execute(named_params! {
                ":deployment_id": deployment_id.to_string(),
                ":component_name": record.component_name.to_string(),
                ":component_type": record.component_type.to_string(),
                ":component_digest": record.component_digest.clone(),
            })
            .map_err(RusqliteError::from)?;
        }
        Ok(())
    }

    fn compute_file_digest(content: &[u8]) -> ContentDigest {
        let hash: [u8; 32] = Sha256::digest(content).into();
        ContentDigest(Digest(hash))
    }

    fn upload_file_tx(
        tx: &Transaction,
        digest: &ContentDigest,
        content: &[u8],
    ) -> Result<(), DbErrorWrite> {
        let actual = Self::compute_file_digest(content);
        if &actual != digest {
            return Err(DbErrorWriteNonRetriable::ValidationFailed(
                format!("uploaded file digest mismatch: expected {digest}, got {actual}").into(),
            )
            .into());
        }
        let size = i64::try_from(content.len()).map_err(|err| DbErrorGeneric::Uncategorized {
            reason: format!("deployment file too large: {err}").into(),
            context: SpanTrace::capture(),
            source: Some(Arc::new(err)),
            loc: Location::caller(),
        })?;
        tx.execute(
            "INSERT INTO t_file (digest, content, size) VALUES (:digest, :content, :size) \
             ON CONFLICT (digest) DO NOTHING",
            rusqlite::named_params! {
                ":digest": digest.to_string(),
                ":content": content,
                ":size": size,
            },
        )
        .map_err(RusqliteError::from)?;
        Ok(())
    }

    fn get_file_tx(
        tx: &Transaction,
        digest: &ContentDigest,
    ) -> Result<Option<Vec<u8>>, DbErrorRead> {
        tx.query_row(
            "SELECT content FROM t_file WHERE digest = :digest",
            rusqlite::named_params! { ":digest": digest.to_string() },
            |row| row.get("content"),
        )
        .optional()
        .map_err(|err| DbErrorRead::from(RusqliteError::from(err)))
    }

    fn missing_digests_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
    ) -> Result<Vec<ContentDigest>, DbErrorRead> {
        tx.prepare(
            "SELECT df.digest \
             FROM t_deployment_file df \
             LEFT JOIN t_file f ON f.digest = df.digest \
             WHERE df.deployment_id = :deployment_id AND f.digest IS NULL \
             ORDER BY df.digest",
        )?
        .query_map(
            rusqlite::named_params! { ":deployment_id": deployment_id.to_string() },
            |row| row.get("digest"),
        )?
        .collect::<Result<Vec<_>, rusqlite::Error>>()
        .map_err(DbErrorRead::from)
    }

    fn list_deployment_files_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
    ) -> Result<Vec<DeploymentFileRecord>, DbErrorRead> {
        tx.prepare(
            // LEFT JOIN: no FK ties t_deployment_file to t_file, so report every referenced file
            // rather than silently dropping one whose metadata row is (unexpectedly) absent.
            "SELECT df.path, df.digest, f.size FROM t_deployment_file df \
             LEFT JOIN t_file f ON f.digest = df.digest \
             WHERE df.deployment_id = :deployment_id \
             ORDER BY df.path, df.digest",
        )?
        .query_map(
            rusqlite::named_params! { ":deployment_id": deployment_id.to_string() },
            |row| {
                let path: String = row.get("path")?;
                let digest: ContentDigest = row.get("digest")?;
                let size = if let Some(size) = row.get::<_, Option<i64>>("size")? {
                    // size is stored as content.len(); a negative value means DB corruption.
                    u64::try_from(size)
                        .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(2, size))?
                } else {
                    warn!(%path, %digest, "deployment file metadata missing from t_file, reporting size 0");
                    0
                };
                Ok(DeploymentFileRecord { path, digest, size })
            },
        )?
        .collect::<Result<Vec<_>, rusqlite::Error>>()
        .map_err(DbErrorRead::from)
    }

    fn activate_deployment_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
        now: DateTime<Utc>,
    ) -> Result<(), DbErrorWrite> {
        // Demote the currently active or enqueued deployment to inactive.
        tx.execute(
            "UPDATE t_deployment SET status = 'inactive' WHERE status IN ('active', 'enqueued')",
            [],
        )
        .map_err(RusqliteError::from)?;
        // Set target deployment to active, recording activation time.
        let rows = tx
            .execute(
                "UPDATE t_deployment SET status = 'active', last_active_at = :now WHERE deployment_id = :deployment_id",
                rusqlite::named_params! {
                    ":now": now,
                    ":deployment_id": deployment_id.to_string(),
                },
            )
            .map_err(RusqliteError::from)?;
        if rows == 0 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(())
    }

    fn enqueue_deployment_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
    ) -> Result<EnqueueOutcome, DbErrorWrite> {
        let status_opt: Option<String> = tx
            .query_row(
                "SELECT status FROM t_deployment WHERE deployment_id = :deployment_id",
                rusqlite::named_params! { ":deployment_id": deployment_id.to_string() },
                |row| row.get(0),
            )
            .optional()
            .map_err(RusqliteError::from)?;
        let status = status_opt.as_deref();
        if status.is_none() {
            return Err(DbErrorWrite::NotFound);
        }
        // Demote any previously enqueued deployment to inactive.
        tx.execute(
            "UPDATE t_deployment SET status = 'inactive' WHERE status = 'enqueued'",
            [],
        )
        .map_err(RusqliteError::from)?;
        // The target is already active: it stays active, only the pending one was cleared.
        if status == Some("active") {
            return Ok(EnqueueOutcome::AlreadyActive);
        }
        // Set target deployment to enqueued.
        let rows = tx
            .execute(
                "UPDATE t_deployment SET status = 'enqueued' WHERE deployment_id = :deployment_id",
                rusqlite::named_params! {
                    ":deployment_id": deployment_id.to_string(),
                },
            )
            .map_err(RusqliteError::from)?;
        if rows == 0 {
            return Err(DbErrorWrite::NotFound);
        }
        Ok(EnqueueOutcome::Enqueued)
    }

    fn get_deployment_tx(
        tx: &Transaction,
        deployment_id: DeploymentId,
    ) -> Result<Option<DeploymentRecord>, DbErrorRead> {
        let Some(record) = tx
            .query_row(
            "SELECT deployment_id, description, digest, created_at, last_active_at, status, deployment_toml, obelisk_version, created_by \
             FROM t_deployment WHERE deployment_id = :deployment_id",
                rusqlite::named_params! { ":deployment_id": deployment_id.to_string() },
                deployment_record_from_row,
            )
            .optional()
            .map_err(|e| DbErrorRead::from(RusqliteError::from(e)))?
        else {
            return Ok(None);
        };
        Self::with_deployment_files_tx(tx, record).map(Some)
    }

    #[cfg(feature = "test")]
    fn get_active_deployment_tx(tx: &Transaction) -> Result<Option<DeploymentRecord>, DbErrorRead> {
        let Some(record) = tx
            .query_row(
            "SELECT deployment_id, description, digest, created_at, last_active_at, status, deployment_toml, obelisk_version, created_by \
             FROM t_deployment WHERE status = 'active' LIMIT 1",
                [],
                deployment_record_from_row,
            )
            .optional()
            .map_err(|e| DbErrorRead::from(RusqliteError::from(e)))?
        else {
            return Ok(None);
        };
        Self::with_deployment_files_tx(tx, record).map(Some)
    }

    fn list_deployments_tx(
        tx: &Transaction,
        pagination: Pagination<Option<DeploymentId>>,
    ) -> Result<Vec<DeploymentRecord>, DbErrorRead> {
        let mut params: Vec<(&'static str, Box<dyn ToSql>)> = vec![];
        let mut sql = String::from(
            "SELECT deployment_id, description, digest, created_at, last_active_at, status, deployment_toml, obelisk_version, created_by \
             FROM t_deployment",
        );

        if let Some(cursor) = pagination.cursor() {
            params.push((":cursor", Box::new(*cursor)));
            write!(
                sql,
                " WHERE deployment_id {rel} :cursor",
                rel = pagination.rel()
            )
            .expect("writing to string");
        }

        let (inner_order, outer_order) = if pagination.is_desc() {
            ("DESC", "")
        } else {
            ("ASC", "DESC")
        };

        write!(
            sql,
            " ORDER BY deployment_id {inner_order} LIMIT {limit}",
            limit = pagination.length()
        )
        .expect("writing to string");

        let final_sql = if outer_order.is_empty() {
            sql
        } else {
            format!("SELECT * FROM ({sql}) AS sub ORDER BY deployment_id {outer_order}")
        };

        let mut result: Vec<DeploymentRecord> = tx
            .prepare(&final_sql)?
            .query_map::<_, &[(&'static str, &dyn ToSql)], _>(
                params
                    .iter()
                    .map(|(k, v)| (*k, v.as_ref()))
                    .collect::<Vec<_>>()
                    .as_ref(),
                deployment_record_from_row,
            )?
            .collect::<Result<Vec<_>, rusqlite::Error>>()
            .map_err(DbErrorRead::from)?;
        for record in &mut result {
            record.files = Self::list_deployment_files_tx(tx, record.deployment_id)?;
        }

        Ok(result)
    }

    fn with_deployment_files_tx(
        tx: &Transaction,
        mut record: DeploymentRecord,
    ) -> Result<DeploymentRecord, DbErrorRead> {
        record.files = Self::list_deployment_files_tx(tx, record.deployment_id)?;
        Ok(record)
    }

    fn pause_execution(
        tx: &Transaction,
        execution_id: &ExecutionId,
        paused_at: DateTime<Utc>,
    ) -> Result<Version, DbErrorWrite> {
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        let mut appending_version = combined_state.get_next_version_fail_if_finished()?;
        debug!("Pausing with {appending_version}");
        if combined_state.reject_locked_activities()? {
            (appending_version, _) = Self::append(
                tx,
                execution_id,
                AppendRequest {
                    created_at: paused_at,
                    event: ExecutionRequest::Unlocked(Unlocked {
                        unlocked_at: paused_at, // does not matter, about to append `Paused` in same tx.
                        reason: "paused".into(),
                    }),
                },
                appending_version,
            )?;
        }
        let (next_version, _notifier) = Self::append(
            tx,
            execution_id,
            AppendRequest {
                created_at: paused_at,
                event: ExecutionRequest::Paused,
            },
            appending_version,
        )?;
        Ok(next_version)
    }

    fn unpause_execution(
        tx: &Transaction,
        execution_id: &ExecutionId,
        paused_at: DateTime<Utc>,
    ) -> Result<Version, DbErrorWrite> {
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        let appending_version = combined_state.get_next_version_fail_if_finished()?;
        debug!("Unpausing with {appending_version}");
        let (next_version, _) = Self::append(
            tx,
            execution_id,
            AppendRequest {
                created_at: paused_at,
                event: ExecutionRequest::Unpaused,
            },
            appending_version,
        )?;
        Ok(next_version)
    }

    fn cancel_workflow(
        tx: &Transaction,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        let combined_state = Self::get_combined_state(tx, execution_id)?;
        if let Some(outcome) = combined_state.cancel_short_circuit() {
            return Ok(outcome);
        }
        Self::append_cancellation_requested(
            tx,
            execution_id,
            cancelled_at,
            &combined_state,
            CancellationFfqnCheck::Required,
        )
    }

    fn append_cancellation_requested(
        tx: &Transaction,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
        combined_state: &CombinedState,
        ffqn_check: CancellationFfqnCheck,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        if matches!(ffqn_check, CancellationFfqnCheck::Required) {
            combined_state.assert_cancellable_workflow_ffqn()?;
        }
        Self::append(
            tx,
            execution_id,
            AppendRequest {
                created_at: cancelled_at,
                event: ExecutionRequest::CancellationRequested,
            },
            combined_state.get_next_version_assert_not_finished(),
        )?;
        Ok(CancelOutcome::Cancelled)
    }

    fn append_activity_cancellation_requested_tx(
        tx: &Transaction,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
        combined_state: &CombinedState,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        match &combined_state.execution_with_state.pending_state {
            PendingState::Finished(finished) => {
                if finished.result_kind
                    == PendingStateFinishedResultKind::Err(
                        PendingStateFinishedError::ExecutionFailure(
                            ExecutionFailureKind::Cancelled,
                        ),
                    )
                {
                    return Ok(CancelOutcome::Cancelled);
                }
                return Ok(CancelOutcome::AlreadyFinished);
            }
            PendingState::Cancelling(_) => return Ok(CancelOutcome::Cancelled),
            _ => {}
        }
        Self::append_cancellation_requested(
            tx,
            execution_id,
            cancelled_at,
            combined_state,
            CancellationFfqnCheck::Skipped,
        )
    }
}

#[async_trait]
impl DbExecutor for SqlitePool {
    #[instrument(level = Level::TRACE, skip(self))]
    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> {
        let execution_ids_versions = self
            .transaction(
                move |conn| {
                    Self::get_pending_by_ffqns(conn, batch_size, pending_at_or_sooner, &ffqns)
                },
                TxType::Other, // read only
                "lock_pending_by_ffqns_get",
            )
            .await
            .map_err(to_generic_error)?;
        if execution_ids_versions.is_empty() {
            Ok(vec![])
        } else {
            debug!("Locking {execution_ids_versions:?}");
            self.transaction(
                move |tx| {
                    let mut locked_execs = Vec::with_capacity(execution_ids_versions.len());
                    // Append lock
                    for (execution_id, version) in &execution_ids_versions {
                        locked_execs.push(Self::lock_single_execution(
                            tx,
                            created_at,
                            &component_id,
                            true,
                            deployment_id,
                            execution_id,
                            run_id,
                            version,
                            executor_id,
                            lock_expires_at,
                            retry_config,
                        )?);
                    }
                    Ok::<_, DbErrorWrite>(locked_execs)
                },
                TxType::MultipleWrites,
                "lock_pending_by_ffqns_one",
            )
            .await
        }
    }

    #[instrument(level = Level::TRACE, skip(self))]
    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> {
        let current_digest = component_id.component_digest.clone();
        let execution_ids_versions = self
            .transaction(
                move |conn| {
                    Self::get_pending_by_ffqns_auto(
                        conn,
                        batch_size,
                        pending_at_or_sooner,
                        &ffqns,
                        &current_digest,
                    )
                },
                TxType::Other,
                "lock_pending_by_ffqns_auto_get",
            )
            .await
            .map_err(to_generic_error)?;
        if execution_ids_versions.is_empty() {
            Ok(vec![])
        } else {
            debug!("Auto-locking {execution_ids_versions:?}");
            self.transaction(
                move |tx| {
                    let mut locked_execs = Vec::with_capacity(execution_ids_versions.len());
                    for (execution_id, version) in &execution_ids_versions {
                        locked_execs.push(Self::lock_single_execution(
                            tx,
                            created_at,
                            &component_id,
                            false,
                            deployment_id,
                            execution_id,
                            run_id,
                            version,
                            executor_id,
                            lock_expires_at,
                            retry_config,
                        )?);
                    }
                    Ok::<_, DbErrorWrite>(locked_execs)
                },
                TxType::MultipleWrites,
                "lock_pending_by_ffqns_auto_one",
            )
            .await
        }
    }

    #[instrument(level = Level::TRACE, skip(self))]
    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> {
        let component_id = component_id.clone();
        let execution_ids_versions = self
            .transaction(
                {
                    let component_id = component_id.clone();
                    move |conn| {
                        Self::get_pending_by_component_input_digest(
                            conn,
                            batch_size,
                            pending_at_or_sooner,
                            &component_id.component_digest,
                        )
                    }
                },
                TxType::Other, // read only
                "lock_pending_by_component_id_get",
            )
            .await
            .map_err(to_generic_error)?;
        if execution_ids_versions.is_empty() {
            Ok(vec![])
        } else {
            debug!("Locking {execution_ids_versions:?}");
            self.transaction(
                move |tx| {
                    let mut locked_execs = Vec::with_capacity(execution_ids_versions.len());
                    // Append lock
                    for (execution_id, version) in &execution_ids_versions {
                        locked_execs.push(Self::lock_single_execution(
                            tx,
                            created_at,
                            &component_id,
                            true,
                            deployment_id,
                            execution_id,
                            run_id,
                            version,
                            executor_id,
                            lock_expires_at,
                            retry_config,
                        )?);
                    }
                    Ok::<_, DbErrorWrite>(locked_execs)
                },
                TxType::MultipleWrites,
                "lock_pending_by_component_id_one",
            )
            .await
        }
    }

    #[cfg(feature = "test")]
    #[instrument(level = Level::DEBUG, skip(self))]
    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> {
        debug!(%execution_id, "lock_one");
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| {
                Self::lock_single_execution(
                    tx,
                    created_at,
                    &component_id,
                    true,
                    deployment_id,
                    &execution_id,
                    run_id,
                    &version,
                    executor_id,
                    lock_expires_at,
                    retry_config,
                )
            },
            TxType::MultipleWrites, // insert + update t_state
            "lock_inner",
        )
        .await
    }

    #[instrument(level = Level::DEBUG, skip(self, req))]
    async fn append(
        &self,
        execution_id: ExecutionId,
        version: Version,
        req: AppendRequest,
    ) -> Result<AppendResponse, DbErrorWrite> {
        debug!(%req, "append");
        trace!(?req, "append");
        let created_at = req.created_at;
        let (version, notifier) = self
            .transaction(
                move |tx| Self::append(tx, &execution_id, req.clone(), version.clone()),
                TxType::MultipleWrites, // insert + update t_state
                "append",
            )
            .await?;
        self.notify_all(vec![notifier], created_at);
        Ok(version)
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_batch_respond_to_parent(
        &self,
        events: AppendEventsToExecution,
        response: AppendResponseToExecution,
        current_time: DateTime<Utc>,
    ) -> Result<AppendBatchResponse, DbErrorWrite> {
        debug!("append_batch_respond_to_parent");
        if events.execution_id == response.parent_execution_id {
            // Pending state would be wrong.
            // This is not a panic because it depends on DB state.
            return Err(DbErrorWrite::NonRetriable(
                DbErrorWriteNonRetriable::ValidationFailed(
                    "Parameters `execution_id` and `parent_execution_id` cannot be the same".into(),
                ),
            ));
        }
        if events.batch.is_empty() {
            error!("Batch cannot be empty");
            return Err(DbErrorWrite::NonRetriable(
                DbErrorWriteNonRetriable::ValidationFailed("batch cannot be empty".into()),
            ));
        }
        let (version, notifiers) = {
            self.transaction(
                move |tx| {
                    let mut version = events.version.clone();
                    let mut notifier_of_child = None;
                    for append_request in &events.batch {
                        let (v, n) = Self::append(
                            tx,
                            &events.execution_id,
                            append_request.clone(),
                            version,
                        )?;
                        version = v;
                        notifier_of_child = Some(n);
                    }

                    let pending_at_parent = Self::append_response(
                        tx,
                        &response.parent_execution_id,
                        JoinSetResponseEventOuter {
                            created_at: response.created_at,
                            event: JoinSetResponseEvent {
                                join_set_id: response.join_set_id.clone(),
                                event: JoinSetResponse::ChildExecutionFinished {
                                    child_execution_id: response.child_execution_id.clone(),
                                    finished_version: response.finished_version.clone(),
                                    result: response.result.clone(),
                                },
                            },
                        },
                    )?;
                    Ok::<_, DbErrorWrite>((
                        version,
                        vec![
                            notifier_of_child.expect("checked that the batch is not empty"),
                            pending_at_parent,
                        ],
                    ))
                },
                TxType::MultipleWrites,
                "append_batch_respond_to_parent",
            )
            .await?
        };
        self.notify_all(notifiers, current_time);
        Ok(version)
    }

    // Supports only one subscriber (executor) per ffqn.
    // A new subscriber replaces the old one, which will eventually time out, which is fine.
    #[instrument(level = Level::TRACE, skip(self, timeout_fut))]
    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>>,
    ) {
        let unique_tag: u64 = rand::random();
        let (sender, mut receiver) = mpsc::channel(1); // senders must use `try_send`
        {
            let mut pending_subscribers = self.0.pending_subscribers.lock().unwrap();
            for ffqn in ffqns.as_ref() {
                pending_subscribers.insert_ffqn(ffqn.clone(), (sender.clone(), unique_tag));
            }
        }
        async {
            let Ok(execution_ids_versions) = self
                .transaction(
                    {
                        let ffqns = ffqns.clone();
                        move |conn| {
                            if let Some(current_digest) = &current_digest {
                                Self::get_pending_by_ffqns_auto(
                                    conn,
                                    1,
                                    pending_at_or_sooner,
                                    ffqns.as_ref(),
                                    current_digest,
                                )
                            } else {
                                Self::get_pending_by_ffqns(
                                    conn,
                                    1,
                                    pending_at_or_sooner,
                                    ffqns.as_ref(),
                                )
                            }
                        }
                    },
                    TxType::Other, // read only
                    "get_pending_by_ffqns",
                )
                .await
            else {
                trace!(
                    "Ignoring get_pending error and waiting in for timeout to avoid executor repolling too soon"
                );
                timeout_fut.await;
                return;
            };
            if !execution_ids_versions.is_empty() {
                trace!("Not waiting, database already contains new pending executions");
                return;
            }
            tokio::select! { // future's liveness: Dropping the loser immediately.
                _ = receiver.recv() => {
                    trace!("Received a notification");
                }
                () = timeout_fut => {
                }
            }
        }.await;
        // Clean up ffqn_to_pending_subscription in any case
        {
            let mut pending_subscribers = self.0.pending_subscribers.lock().unwrap();
            for ffqn in ffqns.as_ref() {
                match pending_subscribers.remove_ffqn(ffqn) {
                    Some((_, tag)) if tag == unique_tag => {
                        // Cleanup OK.
                    }
                    Some(other) => {
                        // Reinsert foreign sender.
                        pending_subscribers.insert_ffqn(ffqn.clone(), other);
                    }
                    None => {
                        // Value was replaced and cleaned up already.
                    }
                }
            }
        }
    }

    // Supports only one subscriber (executor) per component id.
    // A new subscriber replaces the old one, which will eventually time out, which is fine.
    #[instrument(level = Level::DEBUG, skip(self, timeout_fut))]
    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>>,
    ) {
        let unique_tag: u64 = rand::random();
        let (sender, mut receiver) = mpsc::channel(1); // senders must use `try_send`
        {
            let mut pending_subscribers = self.0.pending_subscribers.lock().unwrap();
            pending_subscribers
                .insert_by_component(component_digest.clone(), (sender.clone(), unique_tag));
        }
        async {
            let Ok(execution_ids_versions) = self
                .transaction(
                    {
                        let input_digest = component_digest.clone();
                        move |conn| Self::get_pending_by_component_input_digest(conn, 1, pending_at_or_sooner, &input_digest)
                    },
                    TxType::Other, // read only
                    "get_pending_by_component_input_digest",
                )
                .await
            else {
                trace!(
                    "Ignoring get_pending error and waiting in for timeout to avoid executor repolling too soon"
                );
                timeout_fut.await;
                return;
            };
            if !execution_ids_versions.is_empty() {
                trace!("Not waiting, database already contains new pending executions");
                return;
            }
            tokio::select! { // future's liveness: Dropping the loser immediately.
                _ = receiver.recv() => {
                    trace!("Received a notification");
                }
                () = timeout_fut => {
                }
            }
        }.await;
        // Clean up ffqn_to_pending_subscription in any case
        {
            let mut pending_subscribers = self.0.pending_subscribers.lock().unwrap();

            match pending_subscribers.remove_by_component(component_digest) {
                Some((_, tag)) if tag == unique_tag => {
                    // Cleanup OK.
                }
                Some(other) => {
                    // Reinsert foreign sender.
                    pending_subscribers.insert_by_component(component_digest.clone(), other);
                }
                None => {
                    // Value was replaced and cleaned up already.
                }
            }
        }
    }

    async fn get_last_execution_event(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<ExecutionEvent, DbErrorRead> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| Self::get_last_execution_event(tx, &execution_id),
            TxType::Other, // read only
            "get_last_execution_event",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn append_activity_cancellation_requested(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| {
                let combined_state = Self::get_combined_state(tx, &execution_id)?;
                SqlitePool::append_activity_cancellation_requested_tx(
                    tx,
                    &execution_id,
                    cancelled_at,
                    &combined_state,
                )
            },
            TxType::MultipleWrites,
            "append_activity_cancellation_requested",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn cancel_workflow(
        &self,
        execution_id: &ExecutionId,
        cancelled_at: DateTime<Utc>,
    ) -> Result<CancelOutcome, DbErrorWrite> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| SqlitePool::cancel_workflow(tx, &execution_id, cancelled_at),
            TxType::MultipleWrites,
            "cancel_workflow",
        )
        .await
    }
}

#[async_trait]
impl DbExternalApi for SqlitePool {
    #[instrument(skip(self))]
    async fn get_backtrace(
        &self,
        execution_id: &ExecutionId,
        filter: BacktraceFilter,
    ) -> Result<BacktraceInfo, DbErrorRead> {
        debug!("get_backtrace");
        let execution_id = execution_id.clone();

        self.transaction(
            move |tx| {
                let select = "SELECT component_id, version_min_including, version_max_excluding, wasm_backtrace FROM t_execution_backtrace e \
                                INNER JOIN t_wasm_backtrace w ON e.backtrace_hash = w.backtrace_hash \
                                WHERE execution_id = :execution_id";
                let mut params: Vec<(&'static str, Box<dyn rusqlite::ToSql>)> = vec![(":execution_id", Box::new(execution_id.to_string()))];
                let select = match &filter {
                    BacktraceFilter::Specific(version) =>{
                        params.push((":version", Box::new(version.0)));
                        format!("{select} AND version_min_including <= :version AND version_max_excluding > :version")
                    },
                    BacktraceFilter::First => format!("{select} ORDER BY version_min_including LIMIT 1"),
                    BacktraceFilter::Last => format!("{select} ORDER BY version_min_including DESC LIMIT 1")
                };
                tx
                    .prepare(&select)
                    ?
                    .query_row::<_, &[(&'static str, &dyn ToSql)], _>(
                        params
                            .iter()
                            .map(|(key, value)| (*key, value.as_ref()))
                            .collect::<Vec<_>>()
                            .as_ref(),
                    |row| {
                        Ok(BacktraceInfo {
                            execution_id: execution_id.clone(),
                            component_id: row.get::<_, JsonWrapper<_> >("component_id")?.0,
                            version_min_including: Version::new(row.get::<_, VersionType>("version_min_including")?),
                            version_max_excluding: Version::new(row.get::<_, VersionType>("version_max_excluding")?),
                            wasm_backtrace: row.get::<_, JsonWrapper<_>>("wasm_backtrace")?.0,
                        })
                    },
                ).map_err(DbErrorRead::from)
            },
            TxType::Other, // read only
            "get_last_backtrace",
        ).await
    }

    #[instrument(skip_all)]
    async fn upsert_source_mapping(
        &self,
        component_digest: &ComponentDigest,
        frame_key: &str,
        is_suffix: bool,
        digest: &ContentDigest,
    ) -> Result<(), DbErrorWrite> {
        let component_digest = component_digest.clone();
        let frame_key = frame_key.to_owned();
        let digest = digest.to_string();
        self.transaction(
            move |tx| {
                tx.prepare(
                    "INSERT INTO t_component_source \
                     (component_digest, frame_key, is_suffix, digest) \
                     VALUES (:component_digest, :frame_key, :is_suffix, :digest) \
                     ON CONFLICT (component_digest, frame_key, is_suffix) \
                     DO UPDATE SET digest = excluded.digest",
                )?
                .execute(named_params! {
                    ":component_digest": component_digest,
                    ":frame_key": frame_key,
                    ":is_suffix": is_suffix,
                    ":digest": digest,
                })?;
                Ok(())
            },
            TxType::Other,
            "upsert_source_mapping",
        )
        .await
    }

    #[instrument(skip_all)]
    async fn resolve_source_digest(
        &self,
        component_digest: &ComponentDigest,
        file: &str,
    ) -> Result<Option<ContentDigest>, DbErrorRead> {
        let component_digest = component_digest.clone();
        let file = file.to_owned();
        self.transaction(
            move |tx| {
                let mut stmt = tx.prepare(
                    "SELECT digest \
                     FROM t_component_source \
                     WHERE component_digest = :component_digest \
                       AND ( \
                           (is_suffix = 0 AND frame_key = :file) \
                        OR (is_suffix = 1 AND \
                            substr(:file, length(:file) - length(frame_key) + 1) = frame_key) \
                       )",
                )?;
                let rows: Vec<ContentDigest> = stmt
                    .query_map(
                        named_params! {
                            ":component_digest": component_digest,
                            ":file": file,
                        },
                        |row| row.get("digest"),
                    )?
                    .collect::<Result<_, _>>()?;
                match rows.len() {
                    0 => Ok(None),
                    1 => Ok(Some(rows.into_iter().next().unwrap())),
                    _ => {
                        warn!("Multiple suffix matches for '{file}', returning None");
                        Ok(None)
                    }
                }
            },
            TxType::Other,
            "resolve_source_digest",
        )
        .await
    }

    #[instrument(skip_all)]
    async fn upsert_component_metadata(
        &self,
        records: Vec<ComponentMetadataRecord>,
    ) -> Result<(), DbErrorWrite> {
        self.transaction(
            move |tx| Self::upsert_component_metadata_tx(tx, &records),
            TxType::MultipleWrites,
            "upsert_component_metadata",
        )
        .await
    }

    #[instrument(skip_all)]
    async fn insert_deployment_components(
        &self,
        deployment_id: DeploymentId,
        records: Vec<DeploymentComponentRecord>,
    ) -> Result<(), DbErrorWrite> {
        self.transaction(
            move |tx| Self::insert_deployment_components_tx(tx, deployment_id, &records),
            TxType::MultipleWrites,
            "insert_deployment_components",
        )
        .await
    }

    #[instrument(skip_all)]
    async fn list_deployment_components(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<DeploymentComponentDetail>, DbErrorRead> {
        self.transaction(
            move |tx| {
                let mut stmt = tx.prepare(
                    "SELECT dc.component_name, dc.component_type, dc.component_digest, \
                            cm.imports_json, cm.exports_json, cm.wit \
                     FROM t_deployment_component dc \
                     JOIN t_component_metadata cm ON dc.component_digest = cm.component_digest \
                     WHERE dc.deployment_id = :deployment_id \
                     ORDER BY dc.component_type, dc.component_name",
                )?;
                let rows = stmt
                    .query_map(
                        named_params! { ":deployment_id": deployment_id.to_string() },
                        |row| {
                            deployment_component_detail_from_row(row).map_err(|err| {
                                rusqlite::Error::ToSqlConversionFailure(Box::new(err))
                            })
                        },
                    )?
                    .collect::<Result<Vec<_>, _>>()?;
                Ok(rows)
            },
            TxType::Other,
            "list_deployment_components",
        )
        .await
    }

    #[instrument(skip_all)]
    async fn get_deployment_component_wit(
        &self,
        deployment_id: DeploymentId,
        component_digest: &ComponentDigest,
    ) -> Result<Option<String>, DbErrorRead> {
        let component_digest = component_digest.clone();
        self.transaction(
            move |tx| {
                tx.prepare(
                    "SELECT cm.wit \
                     FROM t_deployment_component dc \
                     JOIN t_component_metadata cm ON dc.component_digest = cm.component_digest \
                     WHERE dc.deployment_id = :deployment_id \
                       AND dc.component_digest = :component_digest \
                     LIMIT 1",
                )?
                .query_row(
                    named_params! {
                        ":deployment_id": deployment_id.to_string(),
                        ":component_digest": component_digest,
                    },
                    |row| row.get(0),
                )
                .optional()
                .map_err(DbErrorRead::from)
            },
            TxType::Other,
            "get_deployment_component_wit",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_executions(
        &self,
        filter: ListExecutionsFilter,
        pagination: ExecutionListPagination,
    ) -> Result<Vec<ExecutionWithState>, DbErrorGeneric> {
        self.transaction(
            move |tx| Self::list_executions(tx, &filter, &pagination),
            TxType::Other, // read only
            "list_executions",
        )
        .await
        .map_err(to_generic_error)
    }

    #[instrument(skip(self))]
    async fn list_execution_events(
        &self,
        execution_id: &ExecutionId,
        pagination: Pagination<VersionType>,
        include_backtrace_id: bool,
    ) -> Result<ListExecutionEventsResponse, DbErrorRead> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| {
                let events = Self::list_execution_events(
                    tx,
                    &execution_id,
                    pagination,
                    include_backtrace_id,
                )?;
                let max_version = Self::get_max_version(tx, &execution_id)?;
                Ok(ListExecutionEventsResponse {
                    events,
                    max_version,
                })
            },
            TxType::Other, // read only
            "get",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_responses(
        &self,
        execution_id: &ExecutionId,
        pagination: Pagination<u32>,
    ) -> Result<ListResponsesResponse, DbErrorRead> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| {
                let responses = Self::list_responses(tx, &execution_id, Some(pagination))?;
                let max_cursor = Self::get_max_response_cursor(tx, &execution_id)?;
                Ok(ListResponsesResponse {
                    responses,
                    max_cursor,
                })
            },
            TxType::Other, // read only
            "list_responses",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_execution_events_responses(
        &self,
        execution_id: &ExecutionId,
        req_since: &Version,
        req_max_length: VersionType,
        req_include_backtrace_id: bool,
        resp_pagination: Pagination<u32>,
    ) -> Result<ExecutionWithStateRequestsResponses, DbErrorRead> {
        let execution_id = execution_id.clone();
        let req_since = req_since.0;
        self.transaction(
            move |tx| {
                let combined_state = Self::get_combined_state(tx, &execution_id)?;
                let events = Self::list_execution_events(
                    tx,
                    &execution_id,
                    Pagination::NewerThan {
                        length: req_max_length
                            .try_into()
                            .expect("req_max_length fits in u16"),
                        cursor: req_since,
                        including_cursor: true,
                    },
                    req_include_backtrace_id,
                )?;
                let responses = Self::list_responses(tx, &execution_id, Some(resp_pagination))?;
                let max_version = Self::get_max_version(tx, &execution_id)?;
                let max_cursor = Self::get_max_response_cursor(tx, &execution_id)?;
                Ok(ExecutionWithStateRequestsResponses {
                    execution_with_state: combined_state.execution_with_state,
                    events,
                    responses,
                    max_version,
                    max_cursor,
                })
            },
            TxType::Other, // read only
            "list_execution_events_responses",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn upgrade_execution_component(
        &self,
        execution_id: &ExecutionId,
        old: &ComponentDigest,
        new: &ComponentDigest,
        reason: ComponentUpgradeReason,
    ) -> Result<(), DbErrorWrite> {
        let execution_id = execution_id.clone();
        let old = old.clone();
        let new = new.clone();
        self.transaction(
            move |tx| {
                Self::upgrade_execution_component_single_write(
                    tx,
                    &execution_id,
                    &old,
                    &new,
                    reason.clone(),
                )
            },
            TxType::Other, // single write
            "upgrade_execution_component",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_logs(
        &self,
        execution_id: &ExecutionId,
        show_derived: bool,
        filter: LogFilter,
        pagination: Pagination<LogCursor>,
    ) -> Result<ListLogsResponse, DbErrorRead> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| Self::list_logs_tx(tx, &execution_id, show_derived, &filter, &pagination),
            TxType::Other, // read only
            "list_logs",
        )
        .await
    }

    #[instrument(skip(self))]
    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> {
        self.transaction(
            move |tx| {
                Self::list_deployment_states(
                    tx,
                    current_time,
                    pagination,
                    include_deployment_toml,
                    execution_counts,
                )
            },
            TxType::Other, // read only
            "list_deployment_states",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn insert_deployment(&self, record: DeploymentRecord) -> Result<(), DbErrorWrite> {
        self.transaction(
            move |tx| Self::insert_deployment_tx(tx, &record),
            TxType::MultipleWrites,
            "insert_deployment",
        )
        .await
    }

    async fn insert_deployment_with_components(
        &self,
        record: DeploymentRecord,
        component_metadata: Vec<ComponentMetadataRecord>,
        deployment_components: Vec<DeploymentComponentRecord>,
    ) -> Result<(), DbErrorWrite> {
        let deployment_id = record.deployment_id;
        self.transaction(
            move |tx| {
                // Parent row first so the t_deployment_component FK is satisfied within the tx.
                Self::insert_deployment_tx(tx, &record)?;
                Self::upsert_component_metadata_tx(tx, &component_metadata)?;
                Self::insert_deployment_components_tx(tx, deployment_id, &deployment_components)
            },
            TxType::MultipleWrites,
            "insert_deployment_with_components",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn missing_digests(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<ContentDigest>, DbErrorRead> {
        self.transaction(
            move |tx| Self::missing_digests_tx(tx, deployment_id),
            TxType::Other,
            "missing_digests",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_deployment_files(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Vec<DeploymentFileRecord>, DbErrorRead> {
        self.transaction(
            move |tx| Self::list_deployment_files_tx(tx, deployment_id),
            TxType::Other,
            "list_deployment_files",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn gc_orphan_files(&self) -> Result<u64, DbErrorWrite> {
        self.transaction(
            move |tx| {
                let deleted = tx
                    .execute(
                        "DELETE FROM t_file WHERE digest NOT IN \
                         (SELECT digest FROM t_deployment_file \
                          UNION SELECT digest FROM t_component_source)",
                        [],
                    )
                    .map_err(RusqliteError::from)?;
                Ok(deleted as u64)
            },
            TxType::MultipleWrites,
            "gc_orphan_files",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn activate_deployment(
        &self,
        deployment_id: DeploymentId,
        now: DateTime<Utc>,
    ) -> Result<(), DbErrorWrite> {
        self.transaction(
            move |tx| Self::activate_deployment_tx(tx, deployment_id, now),
            TxType::MultipleWrites,
            "activate_deployment",
        )
        .await
    }

    async fn enqueue_deployment(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<EnqueueOutcome, DbErrorWrite> {
        self.transaction(
            move |tx| Self::enqueue_deployment_tx(tx, deployment_id),
            TxType::MultipleWrites,
            "enqueue_deployment",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn get_deployment(
        &self,
        deployment_id: DeploymentId,
    ) -> Result<Option<DeploymentRecord>, DbErrorRead> {
        self.transaction(
            move |tx| Self::get_deployment_tx(tx, deployment_id),
            TxType::Other,
            "get_deployment",
        )
        .await
    }

    #[cfg(feature = "test")]
    #[instrument(skip(self))]
    async fn get_active_deployment(&self) -> Result<Option<DeploymentRecord>, DbErrorRead> {
        self.transaction(
            move |tx| Self::get_active_deployment_tx(tx),
            TxType::Other,
            "get_active_deployment",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn get_current_deployment(&self) -> Result<Option<DeploymentRecord>, DbErrorRead> {
        self.transaction(
            move |tx| {
                let Some(record) = tx
                    .query_row(
                    "SELECT deployment_id, description, digest, created_at, last_active_at, status, deployment_toml, obelisk_version, created_by \
                     FROM t_deployment WHERE status IN ('enqueued', 'active') \
                     ORDER BY CASE status WHEN 'enqueued' THEN 0 ELSE 1 END LIMIT 1",
                        [],
                        deployment_record_from_row,
                    )
                    .optional()
                    .map_err(|e| DbErrorRead::from(RusqliteError::from(e)))?
                else {
                    return Ok(None);
                };
                Self::with_deployment_files_tx(tx, record).map(Some)
            },
            TxType::Other,
            "get_current_deployment",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn list_deployments(
        &self,
        pagination: Pagination<Option<DeploymentId>>,
    ) -> Result<Vec<DeploymentRecord>, DbErrorRead> {
        self.transaction(
            move |tx| Self::list_deployments_tx(tx, pagination),
            TxType::Other,
            "list_deployments",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn pause_execution(
        &self,
        execution_id: &ExecutionId,
        paused_at: DateTime<Utc>,
    ) -> Result<AppendResponse, DbErrorWrite> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| SqlitePool::pause_execution(tx, &execution_id, paused_at),
            TxType::MultipleWrites,
            "pause_execution",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn unpause_execution(
        &self,
        execution_id: &ExecutionId,
        unpaused_at: DateTime<Utc>,
    ) -> Result<AppendResponse, DbErrorWrite> {
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| SqlitePool::unpause_execution(tx, &execution_id, unpaused_at),
            TxType::MultipleWrites,
            "unpause_execution",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn pause_delay(&self, delay_id: &DelayId) -> Result<(), DbErrorWrite> {
        let delay_id = delay_id.clone();
        self.transaction(
            move |tx| {
                let (execution_id, join_set_id) = delay_id.split_to_parts();
                let rows_modified = tx.execute(
                    "UPDATE t_delay SET is_paused = 1 \
                     WHERE execution_id = :execution_id AND join_set_id = :join_set_id AND delay_id = :delay_id",
                    named_params! {
                        ":execution_id": execution_id.to_string(),
                        ":join_set_id": join_set_id.to_string(),
                        ":delay_id": delay_id.to_string(),
                    },
                )?;
                if rows_modified == 0 {
                    return Err(DbErrorWrite::NotFound);
                }
                Ok(())
            },
            TxType::Other,
            "pause_delay",
        )
        .await
    }

    #[instrument(skip(self))]
    async fn unpause_delay(&self, delay_id: &DelayId) -> Result<(), DbErrorWrite> {
        let delay_id = delay_id.clone();
        self.transaction(
            move |tx| {
                let (execution_id, join_set_id) = delay_id.split_to_parts();
                let rows_modified = tx.execute(
                    "UPDATE t_delay SET is_paused = 0 \
                     WHERE execution_id = :execution_id AND join_set_id = :join_set_id AND delay_id = :delay_id",
                    named_params! {
                        ":execution_id": execution_id.to_string(),
                        ":join_set_id": join_set_id.to_string(),
                        ":delay_id": delay_id.to_string(),
                    },
                )?;
                if rows_modified == 0 {
                    return Err(DbErrorWrite::NotFound);
                }
                Ok(())
            },
            TxType::Other,
            "unpause_delay",
        )
        .await
    }
}

#[async_trait]
impl Cas for SqlitePool {
    async fn read_blob(&self, digest: &ContentDigest) -> Result<Option<Vec<u8>>, CasError> {
        let digest = digest.clone();
        self.transaction(
            move |tx| Self::get_file_tx(tx, &digest),
            TxType::Other,
            "cas_read_blob",
        )
        .await
        .map_err(|err| CasError::Uncategorized(err.to_string()))
    }

    async fn write_blob(&self, content: &[u8]) -> Result<ContentDigest, CasError> {
        let digest = Self::compute_file_digest(content);
        let content = content.to_vec();
        {
            let digest = digest.clone();
            self.transaction(
                move |tx| Self::upload_file_tx(tx, &digest, &content),
                TxType::MultipleWrites,
                "cas_write_blob",
            )
            .await
            .map_err(|err| CasError::Uncategorized(err.to_string()))?;
        }
        Ok(digest)
    }

    async fn contains_blob(&self, digest: &ContentDigest) -> Result<bool, CasError> {
        Ok(self.read_blob(digest).await?.is_some())
    }
}

#[async_trait]
impl DbConnection for SqlitePool {
    #[instrument(level = Level::DEBUG, skip_all, fields(execution_id = %req.execution_id))]
    async fn create(&self, req: CreateRequest) -> Result<AppendResponse, DbErrorWrite> {
        debug!("create");
        trace!(?req, "create");
        let created_at = req.created_at;
        let (version, notifier) = self
            .transaction(
                move |tx| Self::create_inner(tx, req.clone()),
                TxType::MultipleWrites,
                "create",
            )
            .await?;
        self.notify_all(vec![notifier], created_at);
        Ok(version)
    }

    #[instrument(level = Level::DEBUG, skip(self))]
    async fn get(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<concepts::storage::ExecutionLog, DbErrorRead> {
        trace!("get");
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| Self::get(tx, &execution_id),
            TxType::Other, // read only
            "get",
        )
        .await
    }

    #[instrument(level = Level::DEBUG, skip(self))]
    async fn get_cancelling(&self, batch_size: u32) -> Result<Vec<ExecutionId>, DbErrorRead> {
        self.transaction(
            move |tx| {
                let mut stmt = tx.prepare(
                    "SELECT execution_id FROM t_state WHERE lifecycle = :lifecycle \
                     ORDER BY created_at LIMIT :batch_size",
                )?;
                let rows = stmt
                    .query_map(
                        named_params! {
                            ":lifecycle": LIFECYCLE_CANCELLING,
                            ":batch_size": batch_size,
                        },
                        |row| row.get::<_, String>("execution_id"),
                    )?
                    .collect::<Result<Vec<_>, _>>()?;
                rows.into_iter()
                    .map(|id| {
                        ExecutionId::from_str(&id)
                            .map_err(|_| consistency_rusqlite("invalid t_state.execution_id"))
                            .map_err(DbErrorRead::from)
                    })
                    .collect()
            },
            TxType::Other, // read only
            "get_cancelling",
        )
        .await
    }

    #[instrument(level = Level::DEBUG, skip(self, batch))]
    async fn append_batch(
        &self,
        current_time: DateTime<Utc>,
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
    ) -> Result<AppendBatchResponse, DbErrorWrite> {
        debug!("append_batch");
        trace!(?batch, "append_batch");
        assert!(!batch.is_empty(), "Empty batch request");

        let (version, notifier) = self
            .transaction(
                move |tx| {
                    let mut version = version.clone();
                    let mut notifier = None;
                    for append_request in &batch {
                        let (v, n) =
                            Self::append(tx, &execution_id, append_request.clone(), version)?;
                        version = v;
                        notifier = Some(n);
                    }
                    Ok::<_, DbErrorWrite>((
                        version,
                        notifier.expect("checked that the batch is not empty"),
                    ))
                },
                TxType::MultipleWrites,
                "append_batch",
            )
            .await?;

        self.notify_all(vec![notifier], current_time);
        Ok(version)
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%execution_id, %version))]
    async fn append_batch_create_new_execution(
        &self,
        current_time: DateTime<Utc>,
        batch: Vec<AppendRequest>,
        execution_id: ExecutionId,
        version: Version,
        child_req: Vec<CreateRequest>,
        backtraces: Vec<BacktraceInfo>,
    ) -> Result<AppendBatchResponse, DbErrorWrite> {
        debug!("append_batch_create_new_execution");
        trace!(?batch, ?child_req, "append_batch_create_new_execution");
        assert!(!batch.is_empty(), "Empty batch request");

        let (version, notifiers) = self
            .transaction(
                move |tx| {
                    let mut notifier = None;
                    let mut version = version.clone();
                    for append_request in &batch {
                        let (v, n) =
                            Self::append(tx, &execution_id, append_request.clone(), version)?;
                        version = v;
                        notifier = Some(n);
                    }
                    let mut notifiers = Vec::new();
                    notifiers.push(notifier.expect("checked that the batch is not empty"));

                    for child_req in &child_req {
                        let (_, notifier) = Self::create_inner(tx, child_req.clone())?;
                        notifiers.push(notifier);
                    }
                    Ok::<_, DbErrorWrite>((version, notifiers))
                },
                TxType::MultipleWrites,
                "append_batch_create_new_execution_inner",
            )
            .await?;
        self.notify_all(notifiers, current_time);
        self.transaction_fire_forget(
            move |tx| {
                for backtrace in &backtraces {
                    Self::append_backtrace(tx, backtrace)?;
                }
                Ok::<_, DbErrorWrite>(())
            },
            "append_batch_create_new_execution_append_backtrace",
        )
        .await;
        Ok(version)
    }

    // Supports only one subscriber per execution id.
    // A new call will overwrite the old subscriber, the old one will end
    // with a timeout, which is fine.
    #[instrument(level = Level::DEBUG, skip(self, timeout_fut))]
    async fn subscribe_to_next_responses(
        &self,
        execution_id: &ExecutionId,
        last_response: ResponseCursor,
        timeout_fut: Pin<Box<dyn Future<Output = TimeoutOutcome> + Send>>,
    ) -> Result<Vec<ResponseWithCursor>, DbErrorReadWithTimeout> {
        debug!("next_responses");
        let unique_tag: u64 = rand::random();
        let execution_id = execution_id.clone();

        let cleanup = || {
            let mut guard = self.0.response_subscribers.lock().unwrap();
            match guard.remove(&execution_id) {
                Some((_, tag)) if tag == unique_tag => {} // Cleanup OK.
                Some(other) => {
                    // Reinsert foreign sender.
                    guard.insert(execution_id.clone(), other);
                }
                None => {} // Value was replaced and cleaned up already, or notification was sent.
            }
        };

        let response_subscribers = self.0.response_subscribers.clone();
        let resp_or_receiver = {
            let execution_id = execution_id.clone();
            self.transaction(
                move |tx| {
                    let responses = Self::get_responses_after(tx, &execution_id, last_response)?;
                    if responses.is_empty() {
                        // cannot race as we have the transaction write lock
                        let (sender, receiver) = oneshot::channel();
                        response_subscribers
                            .lock()
                            .unwrap()
                            .insert(execution_id.clone(), (sender, unique_tag));
                        Ok::<_, DbErrorReadWithTimeout>(itertools::Either::Right(receiver))
                    } else {
                        Ok(itertools::Either::Left(responses))
                    }
                },
                TxType::Other, // read only
                "subscribe_to_next_responses",
            )
            .await
        }
        .inspect_err(|_| {
            cleanup();
        })?;
        match resp_or_receiver {
            itertools::Either::Left(resp) => Ok(resp), // no need for cleanup
            itertools::Either::Right(receiver) => {
                let woken = tokio::select! {
                    resp = receiver => match resp {
                        Ok(()) => Ok(()),
                        Err(_) => Err(DbErrorReadWithTimeout::from(DbErrorGeneric::Close)),
                    },
                    outcome = timeout_fut => Err(DbErrorReadWithTimeout::Timeout(outcome)),
                };
                cleanup();
                woken?;
                // Re-fetch from the `last_response` watermark.
                let execution_id = execution_id.clone();
                self.transaction(
                    move |tx| {
                        Self::get_responses_after(tx, &execution_id, last_response)
                            .map_err(DbErrorReadWithTimeout::from)
                    },
                    TxType::Other, // read only
                    "subscribe_to_next_responses_refetch",
                )
                .await
            }
        }
    }

    // Supports multiple subscribers.
    #[instrument(level = Level::DEBUG, skip(self, timeout_fut))]
    async fn wait_for_finished_result(
        &self,
        execution_id: &ExecutionId,
        timeout_fut: Option<Pin<Box<dyn Future<Output = TimeoutOutcome> + Send>>>,
    ) -> Result<SupportedFunctionReturnValue, DbErrorReadWithTimeout> {
        let unique_tag: u64 = rand::random();
        let execution_id = execution_id.clone();
        let execution_finished_subscription = self.0.execution_finished_subscribers.clone();

        let cleanup = || {
            let mut guard = self.0.execution_finished_subscribers.lock().unwrap();
            if let Some(subscribers) = guard.get_mut(&execution_id) {
                subscribers.remove(&unique_tag);
            }
        };

        let resp_or_receiver = {
            let execution_id = execution_id.clone();
            self.transaction(move |tx| {
                let pending_state =
                    Self::get_combined_state(tx, &execution_id)?.execution_with_state.pending_state;
                if let PendingState::Finished(finished) = pending_state {
                    let event =
                        Self::get_execution_event(tx, &execution_id, finished.version)?;
                    if let ExecutionRequest::Finished { retval, ..} = event.event {
                        Ok(itertools::Either::Left(retval))
                    } else {
                        error!("Mismatch, expected Finished row: {event:?} based on t_state {finished}");
                        Err(DbErrorReadWithTimeout::from(consistency_db_err(
                            "cannot get finished event based on t_state version"
                        )))
                    }
                } else {
                    // Cannot race with the notifier as we have the transaction write lock:
                    // Either the finished event was appended previously, thus `itertools::Either::Left` was selected,
                    // or we end up here. If this tx fails, the cleanup will remove this entry.
                    let (sender, receiver) = oneshot::channel();
                    let mut guard = execution_finished_subscription.lock().unwrap();
                    guard.entry(execution_id.clone()).or_default().insert(unique_tag, sender);
                    Ok(itertools::Either::Right(receiver))
                }
            },
            TxType::Other, // read only
            "wait_for_finished_result")
            .await
        }
        .inspect_err(|_| {
            // This cleanup can race with the notification sender, since both are running after a transaction was finished.
            // If the notification sender wins, it removes our oneshot sender and puts a value in it, cleanup will not find the unique tag.
            // If cleanup wins, it simply removes the oneshot sender.
            cleanup();
        })?;

        let timeout_fut = timeout_fut.unwrap_or_else(|| Box::pin(std::future::pending()));
        match resp_or_receiver {
            itertools::Either::Left(resp) => Ok(resp), // no need for cleanup
            itertools::Either::Right(receiver) => {
                let res = tokio::select! {
                    resp = receiver => {
                        match resp {
                            Ok(retval) => Ok(retval),
                            Err(_recv_err) => Err(DbErrorGeneric::Close.into())
                        }
                    }
                    outcome = timeout_fut => Err(DbErrorReadWithTimeout::Timeout(outcome)),
                };
                cleanup();
                res
            }
        }
    }

    #[instrument(level = Level::DEBUG, skip_all, fields(%join_set_id, %execution_id))]
    async fn append_delay_response(
        &self,
        created_at: DateTime<Utc>,
        execution_id: ExecutionId,
        join_set_id: JoinSetId,
        delay_id: DelayId,
        result: Result<(), ()>,
    ) -> Result<AppendDelayResponseOutcome, DbErrorWrite> {
        debug!("append_delay_response");
        let event = JoinSetResponseEventOuter {
            created_at,
            event: JoinSetResponseEvent {
                join_set_id,
                event: JoinSetResponse::DelayFinished {
                    delay_id: delay_id.clone(),
                    result,
                },
            },
        };
        let res = self
            .transaction(
                {
                    let execution_id = execution_id.clone();
                    move |tx| Self::append_response(tx, &execution_id, event.clone())
                },
                TxType::MultipleWrites,
                "append_delay_response",
            )
            .await;
        match res {
            Ok(notifier) => {
                self.notify_all(vec![notifier], created_at);
                Ok(AppendDelayResponseOutcome::Success)
            }
            Err(DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::Conflict)) => {
                let delay_success = self
                    .transaction(
                        move |tx| Self::get_delay_response(tx, &execution_id, &delay_id),
                        TxType::Other, // read only
                        "get_delay_response",
                    )
                    .await?;
                match delay_success {
                    Some(true) => Ok(AppendDelayResponseOutcome::AlreadyFinished),
                    Some(false) => Ok(AppendDelayResponseOutcome::AlreadyCancelled),
                    None => Err(DbErrorWrite::Generic(DbErrorGeneric::Uncategorized {
                        reason: "insert failed yet select did not find the response".into(),
                        context: SpanTrace::capture(),
                        source: None,
                        loc: Location::caller(),
                    })),
                }
            }
            Err(err) => Err(err),
        }
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_backtrace(&self, append: BacktraceInfo) -> Result<(), DbErrorWrite> {
        trace!("append_backtrace");
        self.transaction_fire_forget(
            move |tx| Self::append_backtrace(tx, &append).map(drop),
            "append_backtrace",
        )
        .await;
        Ok(())
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_backtrace_batch(
        &self,
        batch: Vec<BacktraceInfo>,
    ) -> Result<usize, DbErrorWrite> {
        trace!("append_backtrace_batch");
        self.transaction(
            move |tx| {
                let mut inserted = 0;
                for append in &batch {
                    inserted += Self::append_backtrace(tx, append)?;
                }
                Ok::<_, DbErrorWrite>(inserted)
            },
            TxType::MultipleWrites,
            "append_backtrace_batch",
        )
        .await
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_log(&self, row: LogInfoAppendRow) -> Result<(), DbErrorWrite> {
        trace!("append_log");
        self.transaction_fire_forget(move |tx| Self::append_log(tx, &row), "append_log")
            .await;
        Ok(())
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn append_log_batch(&self, batch: &[LogInfoAppendRow]) -> Result<(), DbErrorWrite> {
        trace!("append_log_batch");
        let batch = Vec::from(batch);
        self.transaction_fire_forget(
            move |tx| {
                for row in &batch {
                    Self::append_log(tx, row)?;
                }
                Ok::<_, DbErrorWrite>(())
            },
            "append_log_batch",
        )
        .await;
        Ok(())
    }

    /// Get currently expired delays and locks.
    #[instrument(level = Level::TRACE, skip(self))]
    async fn get_expired_timers(
        &self,
        at: DateTime<Utc>,
    ) -> Result<Vec<ExpiredTimer>, DbErrorGeneric> {
        self.transaction(
            move |conn| {
                let mut expired_timers = conn.prepare(
                    "SELECT execution_id, join_set_id, delay_id FROM t_delay WHERE expires_at <= :at AND NOT is_paused",
                )?
                .query_map(
                        named_params! {
                            ":at": at,
                        },
                        |row| {
                            let execution_id = row.get("execution_id")?;
                            let join_set_id = row.get::<_, JoinSetId>("join_set_id")?;
                            let delay_id = row.get::<_, DelayId>("delay_id")?;
                            let delay = ExpiredDelay { execution_id, join_set_id, delay_id };
                            Ok(ExpiredTimer::Delay(delay))
                        },
                    )?
                    .collect::<Result<Vec<_>, _>>()?;
                // Extend with expired locks
                let expired = conn.prepare(&format!(r#"
                    SELECT execution_id, last_lock_version, corresponding_version, intermittent_event_count, max_retries, retry_exp_backoff_millis,
                    executor_id, run_id, lifecycle
                    FROM t_state
                    WHERE pending_expires_finished <= :at AND state = "{STATE_LOCKED}"
                    "#
                )
                )?
                .query_map(
                        named_params! {
                            ":at": at,
                        },
                        |row| {
                            let execution_id = row.get("execution_id")?;
                            let lifecycle: String = row.get("lifecycle")?;
                            if lifecycle != LIFECYCLE_ACTIVE {
                                error!(%execution_id, %lifecycle, "encountered invalid non-active locked execution while scanning expired locks");
                                return Ok(None);
                            }
                            let locked_at_version = Version::new(row.get("last_lock_version")?);
                            let next_version = Version::new(row.get("corresponding_version")?).increment();
                            let intermittent_event_count = row.get("intermittent_event_count")?;
                            let max_retries = row.get("max_retries")?;
                            let retry_exp_backoff_millis = u64::from(row.get::<_, u32>("retry_exp_backoff_millis")?);
                            let executor_id = row.get("executor_id")?;
                            let run_id = row.get("run_id")?;
                            let lock = ExpiredLock {
                                execution_id,
                                locked_at_version,
                                next_version,
                                intermittent_event_count,
                                max_retries,
                                retry_exp_backoff: Duration::from_millis(retry_exp_backoff_millis),
                                locked_by: LockedBy { executor_id, run_id },
                            };
                            Ok(Some(ExpiredTimer::Lock(lock)))
                        }
                    )?
                    .collect::<Result<Vec<_>, _>>()?;
                expired_timers.extend(expired.into_iter().flatten());
                if !expired_timers.is_empty() {
                    debug!("get_expired_timers found {expired_timers:?}");
                }
                Ok(expired_timers)
            },
            TxType::Other, // read only
            "get_expired_timers"
        )
        .await
        .map_err(to_generic_error)
    }

    async fn get_execution_event(
        &self,
        execution_id: &ExecutionId,
        version: &Version,
    ) -> Result<ExecutionEvent, DbErrorRead> {
        let version = version.0;
        let execution_id = execution_id.clone();
        self.transaction(
            move |tx| Self::get_execution_event(tx, &execution_id, version),
            TxType::Other, // read only
            "get_execution_event",
        )
        .await
    }

    #[instrument(level = Level::DEBUG, skip_all)]
    async fn upsert_stub_response(
        &self,
        execution_id: ExecutionIdDerived,
        version: Version,
        req: AppendRequest,
        response: AppendResponseToExecution,
        current_time: DateTime<Utc>,
    ) -> Result<(), DbErrorStubResponse> {
        debug!("upsert_stub_response");
        #[cfg(debug_assertions)]
        {
            let (expected_parent, expected_join_set) = execution_id.split_to_parts();
            debug_assert_eq!(expected_parent, response.parent_execution_id);
            debug_assert_eq!(expected_join_set, response.join_set_id);
            debug_assert_eq!(execution_id, response.child_execution_id);
        }
        let execution_id = ExecutionId::Derived(execution_id);
        let expected_retval = response.result.clone();
        let notifiers = self
            .transaction(
                move |tx| {
                    let version_raw = version.0;
                    match Self::append(tx, &execution_id, req.clone(), version.clone()) {
                        Ok((_next_version, notifier_of_child)) => {
                            let pending_at_parent = Self::append_response(
                                tx,
                                &response.parent_execution_id,
                                JoinSetResponseEventOuter {
                                    created_at: response.created_at,
                                    event: JoinSetResponseEvent {
                                        join_set_id: response.join_set_id.clone(),
                                        event: JoinSetResponse::ChildExecutionFinished {
                                            child_execution_id: response.child_execution_id.clone(),
                                            finished_version: response.finished_version.clone(),
                                            result: response.result.clone(),
                                        },
                                    },
                                },
                            )
                            .map_err(DbErrorStubResponse::Write)?;
                            Ok::<_, DbErrorStubResponse>(Some(vec![
                                notifier_of_child,
                                pending_at_parent,
                            ]))
                        }
                        Err(DbErrorWrite::NonRetriable(
                            DbErrorWriteNonRetriable::AlreadyFinished,
                        )) => {
                            // Idempotency check: read the existing event and compare retval.
                            let found = Self::get_execution_event(tx, &execution_id, version_raw)
                                .map_err(|_| DbErrorStubResponse::StubConflict)?;
                            match found.event {
                                ExecutionRequest::Finished { retval, .. }
                                    if retval == expected_retval =>
                                {
                                    Ok(None)
                                }
                                _ => Err(DbErrorStubResponse::StubConflict),
                            }
                        }
                        Err(other) => Err(DbErrorStubResponse::Write(other)),
                    }
                },
                TxType::MultipleWrites,
                "upsert_stub_response",
            )
            .await?;
        if let Some(notifiers) = notifiers {
            self.notify_all(notifiers, current_time);
        }
        Ok(())
    }

    async fn get_pending_state(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<ExecutionWithState, DbErrorRead> {
        let execution_id = execution_id.clone();
        Ok(self
            .transaction(
                move |tx| Self::get_combined_state(tx, &execution_id),
                TxType::Other, // read only
                "get_pending_state",
            )
            .await?
            .execution_with_state)
    }
}

#[cfg(feature = "test")]
#[async_trait]
impl concepts::storage::DbConnectionTest for SqlitePool {
    #[instrument(level = Level::DEBUG, skip(self, response_event), fields(join_set_id = %response_event.join_set_id))]
    async fn append_response(
        &self,
        created_at: DateTime<Utc>,
        execution_id: ExecutionId,
        response_event: JoinSetResponseEvent,
    ) -> Result<(), DbErrorWrite> {
        debug!("append_response");
        let event = JoinSetResponseEventOuter {
            created_at,
            event: response_event,
        };
        let notifier = self
            .transaction(
                move |tx| Self::append_response(tx, &execution_id, event.clone()),
                TxType::Other, // read only
                "append_response",
            )
            .await?;
        self.notify_all(vec![notifier], created_at);
        Ok(())
    }
}

#[cfg(any(test, feature = "tempfile"))]
pub mod tempfile {
    use super::{SqliteConfig, SqlitePool};
    use tempfile::NamedTempFile;

    pub async fn sqlite_pool() -> (SqlitePool, Option<NamedTempFile>) {
        if let Ok(path) = std::env::var("SQLITE_FILE") {
            (
                SqlitePool::new(path, SqliteConfig::default())
                    .await
                    .unwrap(),
                None,
            )
        } else {
            let file = NamedTempFile::new().unwrap();
            let path = file.path();
            (
                SqlitePool::new(path, SqliteConfig::default())
                    .await
                    .unwrap(),
                Some(file),
            )
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::sqlite_dao::{SqlitePool, TxType, tempfile::sqlite_pool};
    use assert_matches::assert_matches;
    use chrono::DateTime;
    use concepts::{
        ComponentId, FunctionFqn, Params,
        prefixed_ulid::{DEPLOYMENT_ID_DUMMY, EXECUTION_ID_DUMMY},
        storage::{CreateRequest, DbErrorWrite, DbErrorWriteNonRetriable, DbPoolCloseable},
    };
    use rusqlite::named_params;

    const SOME_FFQN: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn");

    #[tokio::test]
    async fn failing_ltx_should_be_rolled_back() -> Result<(), DbErrorWrite> {
        let created_at = DateTime::from_timestamp_nanos(0);
        let (pool, _guard) = sqlite_pool().await;
        pool.transaction(
            move |tx| {
                let req = CreateRequest {
                    created_at,
                    execution_id: EXECUTION_ID_DUMMY,
                    ffqn: SOME_FFQN,
                    params: Params::empty(),
                    parent: None,
                    metadata: concepts::ExecutionMetadata::empty(),
                    scheduled_at: created_at,
                    component_id: ComponentId::dummy_activity(),
                    deployment_id: DEPLOYMENT_ID_DUMMY,
                    scheduled_by: None,
                    paused: false,
                };
                SqlitePool::create_inner(tx, req)?;
                SqlitePool::pause_execution(tx, &EXECUTION_ID_DUMMY, created_at)?;
                Ok::<_, DbErrorWrite>(())
            },
            TxType::MultipleWrites,
            "create_inner + pause_execution",
        )
        .await?;

        // Second tx should fail
        let err = pool
            .transaction(
                move |tx| SqlitePool::pause_execution(tx, &EXECUTION_ID_DUMMY, created_at),
                TxType::MultipleWrites,
                "pause_execution",
            )
            .await
            .unwrap_err();
        let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
        assert_eq!("cannot pause, execution is already paused", reason.as_ref());

        let events = pool.transaction(
            move |tx| {
                let events =
                    tx.prepare(
                        "SELECT created_at, json_value, version FROM t_execution_log WHERE execution_id = :execution_id",
                    )?
                    .query_map(
                        named_params! {
                            ":execution_id": EXECUTION_ID_DUMMY.to_string(),
                        },
                        SqlitePool::map_t_execution_log_row,
                    )
                    .map_err(DbErrorWrite::from)?
                    .collect::<Result<Vec<_>, _>>()?;

                Ok::<_, DbErrorWrite>(events)
            },
            TxType::Other, // read only
            "get_log",
        )
        .await?;
        assert_eq!(2, events.len());
        pool.close().await;
        Ok(())
    }
}