Trait StateProvider 

pub trait StateProvider: Send + Sync {
    // Required methods
    fn init<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn insert_workflow_status<'life0, 'async_trait>(
        &'life0 self,
        status: WorkflowStatus,
    ) -> Pin<Box<dyn Future<Output = Result<(WorkflowStatus, bool)>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_deduplicated_workflow<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        queue_name: &'life1 str,
        dedup_id: &'life2 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn get_workflow_status<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<WorkflowStatus>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn set_workflow_status<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
        status: &'life2 str,
        output: Option<&'life3 Value>,
        error: Option<&'life4 str>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait,
             'life4: 'async_trait;
    fn get_step_result<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        seq: i32,
    ) -> Pin<Box<dyn Future<Output = Result<Option<RecordedStep>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn record_step_result<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        seq: i32,
        name: &'life2 str,
        value: Value,
        error: Option<&'life3 str>,
        started_at_ms: Option<i64>,
    ) -> Pin<Box<dyn Future<Output = Result<StepOutcome>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn run_transaction_step<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        seq: i32,
        started_at_ms: i64,
        opts: &'life2 TransactionOptions,
        body: TxBody<'life3>,
    ) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn dequeue_workflows<'life0, 'life1, 'async_trait>(
        &'life0 self,
        req: &'life1 DequeueRequest,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowStatus>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn transition_delayed_workflows<'life0, 'async_trait>(
        &'life0 self,
        now_ms: i64,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn queue_partitions<'life0, 'life1, 'async_trait>(
        &'life0 self,
        queue_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn insert_notification<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        destination_id: &'life1 str,
        topic: &'life2 str,
        message: Value,
        idempotency_key: Option<&'life3 str>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn consume_notification<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        topic: &'life2 str,
        seq: i32,
        step_name: &'life3 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Value>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn upsert_event<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        key: &'life2 str,
        value: Value,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn get_event_value<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        key: &'life2 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Value>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn list_workflows<'life0, 'life1, 'async_trait>(
        &'life0 self,
        filter: &'life1 ListFilter,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowStatus>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_workflow_aggregates<'life0, 'life1, 'async_trait>(
        &'life0 self,
        query: &'life1 WorkflowAggregateQuery,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowAggregate>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_step_aggregates<'life0, 'life1, 'async_trait>(
        &'life0 self,
        query: &'life1 StepAggregateQuery,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<StepAggregate>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn cancel_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn resume_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn enqueue_existing<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
        queue: &'life2 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn cancel_workflows<'life0, 'life1, 'async_trait>(
        &'life0 self,
        ids: &'life1 [String],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn resume_workflows<'life0, 'life1, 'async_trait>(
        &'life0 self,
        ids: &'life1 [String],
    ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn delete_workflows<'life0, 'life1, 'async_trait>(
        &'life0 self,
        ids: &'life1 [String],
        delete_children: bool,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn set_workflow_delay<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
        delay_until_ms: i64,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn fork_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        params: &'life1 ForkParams,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn bump_recovery_attempts<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
        max: i32,
    ) -> Pin<Box<dyn Future<Output = Result<i32>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn record_child_workflow<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        parent_id: &'life1 str,
        seq: i32,
        name: &'life2 str,
        child_id: &'life3 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn check_child_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        parent_id: &'life1 str,
        seq: i32,
    ) -> Pin<Box<dyn Future<Output = Result<Option<(String, String)>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_workflow_steps<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<StepInfo>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_step_name<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        seq: i32,
    ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn record_patch<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        seq: i32,
        name: &'life2 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn write_stream<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        key: &'life2 str,
        value: Option<Value>,
        function_id: i32,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn read_stream<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        key: &'life2 str,
        from_offset: i32,
    ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, bool)>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn list_workflow_events<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Value)>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_workflow_notifications<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<NotificationInfo>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_workflow_streams<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Vec<Value>)>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn create_schedule<'life0, 'life1, 'async_trait>(
        &'life0 self,
        schedule: &'life1 WorkflowSchedule,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn apply_schedules<'life0, 'life1, 'async_trait>(
        &'life0 self,
        schedules: &'life1 [WorkflowSchedule],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_schedules<'life0, 'life1, 'async_trait>(
        &'life0 self,
        filter: &'life1 ScheduleFilter,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowSchedule>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn set_schedule_status<'life0, 'life1, 'async_trait>(
        &'life0 self,
        name: &'life1 str,
        status: ScheduleStatus,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn set_schedule_last_fired<'life0, 'life1, 'async_trait>(
        &'life0 self,
        name: &'life1 str,
        at_ms: i64,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn delete_schedule<'life0, 'life1, 'async_trait>(
        &'life0 self,
        name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn create_application_version<'life0, 'life1, 'async_trait>(
        &'life0 self,
        version_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_application_versions<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<VersionInfo>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_latest_application_version<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<VersionInfo>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn set_latest_application_version<'life0, 'life1, 'async_trait>(
        &'life0 self,
        version_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn upsert_queue<'life0, 'life1, 'async_trait>(
        &'life0 self,
        queue: &'life1 WorkflowQueue,
        update_existing: bool,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_queues<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowQueue>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn export_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflow_id: &'life1 str,
        export_children: bool,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ExportedWorkflow>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn import_workflow<'life0, 'life1, 'async_trait>(
        &'life0 self,
        workflows: &'life1 [ExportedWorkflow],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;

    // Provided methods
    fn serializer(&self) -> Serializer { ... }
    fn supports_listen_notify(&self) -> bool { ... }
    fn await_change<'life0, 'life1, 'async_trait>(
        &'life0 self,
        wait: ChangeWait<'life1>,
        within: Duration,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
}

The pluggable durable-state backend.

This is the single seam that decouples the runtime from storage. v0.1 ships a Postgres implementation and an in-memory one; a DynamoDB / Aurora DSQL implementation can be added later without touching the engine.

Every method must be idempotent with respect to its keys, because the engine may re-run a workflow after a crash and replay completed steps.

Required Methods

fn init<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create tables / indexes if they do not yet exist.

fn insert_workflow_status<'life0, 'async_trait>( &'life0 self, status: WorkflowStatus, ) -> Pin<Box<dyn Future<Output = Result<(WorkflowStatus, bool)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Idempotently insert a workflow row. If status.id already exists, the existing row is returned unchanged (so a re-submitted id is a no-op, not a duplicate). This is the single creation path for both direct runs and enqueues.

The returned flag is true iff this call created the row — the atomic arbiter for “who runs it” when several executors race the same deterministic id (e.g. a scheduled tick). Executor ids cannot serve as that arbiter: they are not unique across processes ("local" is the default for every process not told otherwise).

fn get_deduplicated_workflow<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queue_name: &'life1 str, dedup_id: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The id of the workflow currently holding the deduplication slot (queue_name, dedup_id), if any. Backs DeduplicationPolicy::ReturnExisting: on a dedup collision the enqueue returns a handle to this workflow.

fn get_workflow_status<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<WorkflowStatus>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Fetch a workflow row by id, if it exists.

fn set_workflow_status<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, id: &'life1 str, status: &'life2 str, output: Option<&'life3 Value>, error: Option<&'life4 str>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Transition a workflow to a new status, optionally writing its terminal output or error. Bumps updated_at.

fn get_step_result<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, seq: i32, ) -> Pin<Box<dyn Future<Output = Result<Option<RecordedStep>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Return a previously checkpointed step — its recorded name plus outcome (output or recorded failure) — or None if the step has not run. The name lets the replayer detect a non-deterministic workflow (a different operation recorded at this position than the one now executing).

fn record_step_result<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, workflow_id: &'life1 str, seq: i32, name: &'life2 str, value: Value, error: Option<&'life3 str>, started_at_ms: Option<i64>, ) -> Pin<Box<dyn Future<Output = Result<StepOutcome>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Idempotently record a step’s outcome keyed by (workflow_id, seq): its success value, or — when error is set — its already-encoded failure (stored in the error column, output left null). A step records exactly one of the two.

Returns the canonical stored outcome: if a concurrent/duplicate execution already wrote this step, the previously-stored outcome wins and is returned, guaranteeing every caller observes the same result (so a step that another execution recorded as failed is observed as failed, and vice versa).

started_at_ms is when the step’s work began (epoch ms); the implementation stamps completed_at itself as the time of the write. None records no start time — used for instantaneous operations (sends, event sets, sleep markers) that have no duration; such rows are excluded from step duration aggregates.

Durable sleep is built on this too: the wake instant is recorded as an ordinary step (DBOS.sleep) in operation_outputs — there is no separate timers table.

fn run_transaction_step<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, workflow_id: &'life1 str, seq: i32, started_at_ms: i64, opts: &'life2 TransactionOptions, body: TxBody<'life3>, ) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Run a transactional step: body’s SQL writes and this step’s operation_outputs checkpoint commit in one database transaction, so the writes happen exactly once. Returns the step’s recorded outcome — its output as Ok (body’s on the first run, the stored one on replay), or a recorded failure as Err.

On a body error the body’s writes roll back (the step stays atomic), but the error is then recorded outside the aborted transaction, so the failed step is durable: a replay returns the recorded error without re-running body (like an ordinary step). A transaction-level conflict (serialization failure / deadlock) is not recorded — it restarts the whole transaction on a fresh one, re-running body. SQL backends only; the in-memory provider returns an error.

fn dequeue_workflows<'life0, 'life1, 'async_trait>( &'life0 self, req: &'life1 DequeueRequest, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowStatus>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically claim up to req.max_tasks ENQUEUED workflows from a queue, transitioning them to PENDING stamped with the claiming executor, the app version, and started_at. Candidates are ordered by (priority, created_at). Honors global_concurrency (queue-wide PENDING cap) and the rate-limit window if set; for workflows with a stored timeout_ms, the absolute deadline is fixed at claim time.

Must be safe under concurrent dispatchers: a workflow is claimed by exactly one caller (Postgres uses FOR UPDATE SKIP LOCKED / NOWAIT).

fn transition_delayed_workflows<'life0, 'async_trait>( &'life0 self, now_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Transition every DELAYED workflow whose delay_until_ms <= now_ms to ENQUEUED. Returns how many were transitioned. Called by the dispatcher at the top of each polling iteration.

fn queue_partitions<'life0, 'life1, 'async_trait>( &'life0 self, queue_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Distinct non-null partition keys among the ENQUEUED workflows on queue_name. The dispatcher of a partitioned queue iterates these and dequeues each partition independently.

fn insert_notification<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, destination_id: &'life1 str, topic: &'life2 str, message: Value, idempotency_key: Option<&'life3 str>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Append a message for destination_id on topic. Errors if the destination workflow does not exist (FK violation in the SQL backends).

When idempotency_key is Some, the row’s primary key is derived from it ({key}::{destination_id}) and a repeated insert is a silent no-op, so a caller that retries the send delivers the message at most once; None assigns a fresh id, so every send delivers.

fn consume_notification<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, workflow_id: &'life1 str, topic: &'life2 str, seq: i32, step_name: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Atomically claim the oldest unconsumed message for (workflow_id, topic) and record it as the step checkpoint (workflow_id, seq) in the same transaction — if claiming and checkpointing were separate, a crash between them would lose the message. Returns the message, or None when the mailbox is empty (nothing is recorded in that case).

fn upsert_event<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 str, key: &'life2 str, value: Value, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Set (or overwrite) the value of event key on workflow_id.

fn get_event_value<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 str, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read the current value of event key on workflow_id, if set.

fn list_workflows<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ListFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowStatus>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List workflows matching filter, newest- or oldest-first per filter.sort_desc.

fn get_workflow_aggregates<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 WorkflowAggregateQuery, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowAggregate>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Count workflows grouped per query (one WorkflowAggregate per non-empty group). The engine validates that the query groups by at least one dimension before calling this.

fn get_step_aggregates<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 StepAggregateQuery, ) -> Pin<Box<dyn Future<Output = Result<Vec<StepAggregate>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Aggregate step (operation_outputs) rows grouped per query, selecting count and/or max duration. The engine validates that the query groups by at least one dimension and selects at least one aggregate before calling.

fn cancel_workflow<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Cancel a workflow: if it is not already terminal, set it CANCELLED, stamp completed_at, and clear queue assignment / dedup so it leaves any queue. A running workflow stops cooperatively at its next step.

fn resume_workflow<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resume a CANCELLED (or otherwise non-terminal) workflow by returning it to PENDING, resetting recovery_attempts and clearing deadline / dedup / started / completed. Returns true if a row was actually transitioned (i.e. it existed and was not already SUCCESS/ERROR). The caller re-dispatches it.

fn enqueue_existing<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, queue: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Route an existing row to a queue: set it ENQUEUED on queue, clearing the owning executor and start time so a dispatcher claims it fresh. Used to re-execute a resumed workflow on a running engine without re-running it locally. A no-op if the id is gone.

fn cancel_workflows<'life0, 'life1, 'async_trait>( &'life0 self, ids: &'life1 [String], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Cancel many workflows in one round-trip. Each existing, non-terminal id is set CANCELLED (same effect as cancel_workflow); missing or already-terminal ids are silently skipped (no error). An empty slice is a no-op.

fn resume_workflows<'life0, 'life1, 'async_trait>( &'life0 self, ids: &'life1 [String], ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resume many workflows in one round-trip. Each existing id that is not SUCCESS/ERROR returns to PENDING (same effect as resume_workflow). Returns the ids actually transitioned, so the caller can re-dispatch exactly those. An empty slice returns an empty vec.

fn delete_workflows<'life0, 'life1, 'async_trait>( &'life0 self, ids: &'life1 [String], delete_children: bool, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete workflows and (via ON DELETE CASCADE) their step / event / stream rows, regardless of state. When delete_children, every descendant by parent_workflow_id (transitively) is deleted too. Missing ids are skipped. An empty slice is a no-op.

fn set_workflow_delay<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, delay_until_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Reschedule a DELAYED workflow: set its delay_until to delay_until_ms. Only affects a row currently in DELAYED (a queue dispatcher promotes it to ENQUEUED once due); anything else is a no-op. Returns whether a row was updated.

fn fork_workflow<'life0, 'life1, 'async_trait>( &'life0 self, params: &'life1 ForkParams, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create params.new_id as a fork of params.original_id: a fresh ENQUEUED workflow on params.queue_name with the same name/input/auth/class/config/app-id, forked_from = original_id, and the original’s step checkpoints with seq < start_step copied in so execution resumes from that step. Marks the original was_forked_from. Errors if the original does not exist.

fn bump_recovery_attempts<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, max: i32, ) -> Pin<Box<dyn Future<Output = Result<i32>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically increment a workflow’s recovery_attempts and return the new value. If it exceeds max, the workflow is parked in MAX_RECOVERY_ATTEMPTS_EXCEEDED instead of being recovered again.

fn record_child_workflow<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, parent_id: &'life1 str, seq: i32, name: &'life2 str, child_id: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Idempotently record that parent_id’s step seq started child workflow child_id. Stored as an operation_outputs checkpoint carrying the child id, so a replay of the parent re-attaches to the same child instead of starting a new one. A second record for the same (parent_id, seq) is a no-op.

fn check_child_workflow<'life0, 'life1, 'async_trait>( &'life0 self, parent_id: &'life1 str, seq: i32, ) -> Pin<Box<dyn Future<Output = Result<Option<(String, String)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Return (child_id, recorded_name) for the child workflow parent_id started at step seq, if one was recorded by record_child_workflow. The recorded name lets the replayer detect a non-deterministic parent (a different child recorded at this position than the one now being started).

fn get_workflow_steps<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<StepInfo>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List a workflow’s recorded operations (its operation_outputs rows) as StepInfo, ordered by step_id. Outputs are decoded per each row’s recorded serialization format. Returns an empty list for an unknown workflow or one that has run no steps.

fn get_step_name<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, seq: i32, ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The function_name recorded at (workflow_id, seq), if a row exists. Used by the patch system to tell a marker from a pre-patch step.

fn record_patch<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 str, seq: i32, name: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Idempotently record a name-only marker row at (workflow_id, seq) (no output) — the checkpoint the patch system writes so a replay observes the same patch decision. A second record for the same key is a no-op.

fn write_stream<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 str, key: &'life2 str, value: Option<Value>, function_id: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Append one entry to the append-only stream (workflow_id, key) at the next offset (MAX(offset) + 1, starting at 0), stamped with the producing step’s function_id. value is the user value to encode and store; None writes the close sentinel instead, which seals the stream. Errors if the stream is already closed. The destination workflow’s existence is enforced by the streams foreign key.

fn read_stream<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 str, key: &'life2 str, from_offset: i32, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, bool)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read stream (workflow_id, key) entries with offset >= from_offset in order, decoding each per its stored serialization. Returns the decoded values and whether the close sentinel was reached (the sentinel itself is not included). Reading never blocks — the caller polls.

fn list_workflow_events<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Value)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

All (key, value) events set on a workflow (set_event), decoded per their stored serialization, ordered by key. For observability/control planes that surface a workflow’s events.

fn list_workflow_notifications<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<NotificationInfo>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

All notifications sent to a workflow (its send/recv mailbox), oldest first, with each message decoded. Includes already-consumed entries.

fn list_workflow_streams<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Vec<Value>)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

All of a workflow’s streams, grouped by key and ordered by write offset, with values decoded and the close sentinel excluded.

fn create_schedule<'life0, 'life1, 'async_trait>( &'life0 self, schedule: &'life1 WorkflowSchedule, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Insert a schedule row. The schedule_name is unique, so creating one that already exists is a unique violation.

fn apply_schedules<'life0, 'life1, 'async_trait>( &'life0 self, schedules: &'life1 [WorkflowSchedule], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically replace each named schedule (delete-by-name then insert) in a single transaction, so the whole batch is all-or-nothing: a failure on any entry leaves every prior entry — and any pre-existing rows the batch would have replaced — untouched. The caller validates the entries and mints a fresh schedule_id for each beforehand.

fn list_schedules<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ScheduleFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowSchedule>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

All schedules matching filter (empty filter returns every schedule), ordered by schedule_name.

fn set_schedule_status<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, status: ScheduleStatus, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Set a schedule’s status. Returns whether a row matched.

fn set_schedule_last_fired<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, at_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Stamp last_fired_at (epoch ms) on a schedule. A no-op if it is gone.

fn delete_schedule<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a schedule by name. Returns whether a row was removed.

fn create_application_version<'life0, 'life1, 'async_trait>( &'life0 self, version_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Register an application version, idempotently (no-op if version_name already exists). Stamps both timestamps with now.

fn list_application_versions<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<VersionInfo>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

All registered application versions, newest version_timestamp first.

fn get_latest_application_version<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<VersionInfo>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The version with the most recent version_timestamp, or None if none are registered.

fn set_latest_application_version<'life0, 'life1, 'async_trait>( &'life0 self, version_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a version as latest by bumping its version_timestamp to now. Returns whether a row matched (no-op if the name is unknown).

fn upsert_queue<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 WorkflowQueue, update_existing: bool, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist a queue’s configuration into the queues table — the database-backed registry the conductor reads fleet-wide, distinct from the engine’s in-process registry. Keyed by name: a first write inserts; a name collision does nothing unless update_existing, which overwrites the stored configuration. Called once per registered queue on launch.

fn list_queues<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowQueue>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Every queue persisted in the queues table, sorted by name — the database-backed (fleet-wide) counterpart to the engine’s in-process list_registered_queues. Fields not stored in the table (max_tasks_per_iteration, max_polling_interval) come back at their defaults.

fn export_workflow<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 str, export_children: bool, ) -> Pin<Box<dyn Future<Output = Result<Vec<ExportedWorkflow>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Export a workflow and (when export_children) all of its transitive children into the portable ExportedWorkflow form. The root workflow is first in the returned list, followed by descendants discovered through parent_workflow_id. Errors if the root workflow does not exist.

fn import_workflow<'life0, 'life1, 'async_trait>( &'life0 self, workflows: &'life1 [ExportedWorkflow], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Import previously export_workflowed workflows, re-inserting each one’s workflow_status row and dependent rows. Atomic: either every workflow is imported or none is. A workflow whose id already exists is an error (import does not overwrite).

Provided Methods

fn serializer(&self) -> Serializer

The serialization format this provider stores values in. The engine reads it to encode a failed workflow’s error in that same format — errors are encoded at the engine because they carry a structured type the set_workflow_status &str channel cannot — so a portable provider writes the cross-language error envelope. Defaults to Serializer::Json (bare error strings); the SQL providers return their configured serializer.

fn supports_listen_notify(&self) -> bool

Whether this backend pushes change signals (Postgres LISTEN/NOTIFY), so a blocked recv/get_event is woken as soon as the row it waits for is written rather than only by polling. Callers that get true can wait on a long backstop interval and rely on await_change for promptness; false means they must poll at a short interval.

fn await_change<'life0, 'life1, 'async_trait>( &'life0 self, wait: ChangeWait<'life1>, within: Duration, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Wait up to within for a hint that wait’s condition may have changed, returning early when a matching change is signalled. The wake is only a hint: the caller must re-check the database (a signal can be missed in the gap between the caller’s last check and subscribing — the bounded within is the backstop). Backends without push signalling just sleep.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl StateProvider for InMemoryProvider

impl StateProvider for PostgresProvider

Available on crate feature postgres only.

impl StateProvider for SqliteProvider

Available on crate feature sqlite only.