Struct WorkflowContext 

#[non_exhaustive]
pub struct WorkflowContext {
    pub run_id: Uuid,
    pub workflow_name: String,
    pub started_at: DateTime<Utc>,
    pub auth: AuthContext,
    /* private fields */
}

Context available to workflow handlers.

Fields (Non-exhaustive)

Non-exhaustive structs could have additional fields added in future. Therefore, non-exhaustive structs cannot be constructed in external crates using the traditional Struct { .. } syntax; cannot be matched against without a wildcard ..; and struct update syntax will not work.

run_id: Uuid

Workflow run ID.

workflow_name: String

Workflow name.

started_at: DateTime<Utc>

When the workflow started.

auth: AuthContext

Authentication context.

Implementations

impl WorkflowContext

pub fn new( run_id: Uuid, workflow_name: String, db_pool: PgPool, http_client: CircuitBreakerClient, ) -> Self

Create a new workflow context.

pub fn with_persist_step_start(self, persist: bool) -> Self

Enable DB writes for record_step_start. By default, only record_step_complete writes to the database (its upsert handles the missing start row). Enable this for long-running steps where observing in-progress state is important.

pub fn resumed( run_id: Uuid, workflow_name: String, started_at: DateTime<Utc>, db_pool: PgPool, http_client: CircuitBreakerClient, ) -> Self

Create a resumed workflow context.

pub fn with_kv(self, kv: Arc<dyn KvHandle>) -> Self

Attach a KV store handle. Called by the runtime before handing the context to the handler.

pub fn kv(&self) -> Result<&dyn KvHandle>

Access the KV store.

pub fn with_env_provider(self, provider: Arc<dyn EnvProvider>) -> Self

Set environment provider.

pub fn with_resumed_from_sleep(self) -> Self

Mark that this context resumed from a sleep (timer expired).

pub fn with_tenant(self, tenant_id: Uuid) -> Self

Set the tenant ID.

pub fn tenant_id(&self) -> Option<Uuid>

pub fn is_resumed(&self) -> bool

pub fn workflow_time(&self) -> DateTime<Utc>

pub fn db(&self) -> ForgeDb

pub fn db_conn(&self) -> DbConn<'_>

Get a DbConn for use in shared helper functions.

pub async fn conn(&self) -> Result<ForgeConn<'static>>

Acquire a connection compatible with sqlx compile-time checked macros.

pub fn http(&self) -> HttpClient

pub fn raw_http(&self) -> &Client

pub fn set_http_timeout(&mut self, timeout: Option<Duration>)

pub fn with_auth(self, auth: AuthContext) -> Self

Set authentication context.

pub fn with_saved_state(self, state: HashMap<String, Value>) -> Self

Restore saved state from persisted data (used on resume).

pub fn save_state(&self, key: &str, value: impl Serialize) -> Result<()>

Save arbitrary state that persists across suspension points.

pub fn load_state<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>>

Load previously saved state. Returns None if the key doesn’t exist.

pub fn take_saved_state(&self) -> HashMap<String, Value>

Get a snapshot of all saved state for persistence.

pub fn with_step_states(self, states: HashMap<String, StepState>) -> Self

Restore step states from persisted data.

pub fn get_step_state(&self, name: &str) -> Option<StepState>

pub fn is_step_completed(&self, name: &str) -> bool

pub fn is_step_started(&self, name: &str) -> bool

Check if a step has been started (running, completed, or failed).

Use this to guard steps that should only execute once, even across workflow suspension and resumption.

pub fn get_step_result<T: DeserializeOwned>(&self, name: &str) -> Option<T>

pub async fn record_step_start(&self, name: &str) -> Result<()>

Record step start and persist to the database before returning.

If the step is already running or beyond (completed/failed), this is a no-op.

name is part of the workflow’s persisted contract. The #[workflow] macro hashes every step name (along with wait keys, timeout, and type names) into a signature stored at run creation. If you rename a step, the next deploy produces a different signature, and any in-flight run that tries to resume will be blocked with WorkflowStatus::BlockedSignatureMismatch. Treat step names as stable public identifiers — change them only under a new workflow version.

Persistence errors are propagated. A swallowed error here would let the workflow continue running while its on-disk state diverged from memory, producing a “completed” run with no recorded step rows.

pub async fn record_step_complete( &self, name: &str, result: Value, ) -> Result<()>

Record step completion and persist to the database before returning.

Errors from the persist call are propagated so the workflow can react rather than continuing past a step the database never observed.

pub async fn record_step_failure( &self, name: &str, error: impl Into<String>, ) -> Result<()>

Record step failure and persist to the database before returning.

Errors from the persist call are propagated so the workflow doesn’t declare a step “failed” only in memory while the row still claims it is running.

pub async fn record_step_compensated(&self, name: &str) -> Result<()>

Record step compensation and persist to the database before returning.

Persistence is inline (not tokio::spawn’d): if the process crashes after the in-memory state changes but before the row update lands, a later resume would see the step as still completed and re-run its compensation handler. Inline await ties durability to the caller and surfaces failures through Result.

pub fn completed_steps_reversed(&self) -> Vec<String>

pub fn all_step_states(&self) -> HashMap<String, StepState>

pub fn elapsed(&self) -> Duration

pub fn register_compensation( &self, step_name: &str, handler: CompensationHandler, )

Register a compensation handler for a step.

Limitation: compensation handlers are in-memory closures and cannot survive a process restart. If the process crashes between step completion and workflow termination, compensation handlers for completed steps are lost. The WorkflowExecutor::cancel method detects this and fails the workflow with a clear message indicating manual remediation is required. This is an inherent constraint of closure-based compensation; a durable alternative would require serializable compensation descriptors (e.g., naming a registered handler + captured args as JSON).

pub fn get_compensation_handler( &self, step_name: &str, ) -> Option<CompensationHandler>

pub fn has_compensation(&self, step_name: &str) -> bool

pub async fn run_compensation(&self) -> Vec<(String, bool)>

Run compensation for all completed steps in reverse order. Returns a list of (step_name, success) tuples.

pub fn compensation_handlers(&self) -> HashMap<String, CompensationHandler>

pub async fn sleep(&self, duration: Duration) -> Result<()>

Sleep for a duration.

This suspends the workflow and persists the wake time to the database. The workflow scheduler will resume the workflow when the time arrives.

Example

// Sleep for 30 days
ctx.sleep(Duration::from_secs(30 * 24 * 60 * 60)).await?;

pub async fn sleep_until(&self, wake_at: DateTime<Utc>) -> Result<()>

Sleep until a specific time.

If the wake time has already passed, returns immediately.

Example

use chrono::{Utc, Duration};
let renewal_date = Utc::now() + Duration::days(30);
ctx.sleep_until(renewal_date).await?;

pub async fn wait_for_event<T: DeserializeOwned>( &self, event_name: &str, timeout: Option<Duration>, ) -> Result<T>

Wait for an external event with optional timeout.

The workflow suspends until the event arrives or the timeout expires. Events are correlated by the workflow run ID.

Example

let payment: PaymentConfirmation = ctx.wait_for_event(
    "payment_confirmed",
    Some(Duration::from_secs(7 * 24 * 60 * 60)), // 7 days
).await?;

pub fn take_suspend_reason(&self) -> Option<SuspendReason>

Take the stored suspension reason, if any.

Called by the executor after the handler returns an error to determine whether the error represents a suspension or a real failure.

Trait Implementations

impl EnvAccess for WorkflowContext

fn env_provider(&self) -> &dyn EnvProvider

fn env(&self, key: &str) -> Option<String>

fn env_or(&self, key: &str, default: &str) -> String

fn env_require(&self, key: &str) -> Result<String>

fn env_parse<T: FromStr>(&self, key: &str) -> Result<T>

where T::Err: Display,

fn env_parse_or<T: FromStr>(&self, key: &str, default: T) -> Result<T>

where T::Err: Display,

Returns the default when unset; errors only if the variable is set but unparseable.

fn env_contains(&self, key: &str) -> bool

impl HandlerContext for WorkflowContext

fn db(&self) -> ForgeDb

Database handle with automatic db.query tracing spans.

fn db_conn(&self) -> DbConn<'_>

Unified connection handle for shared helper functions.