Struct WorkflowRuntime 

pub struct WorkflowRuntime<S>
where
    S: Store + WorkflowQueryStore,
{ /* private fields */ }

Effect execution runtime.

The runtime coordinates effect and timer workers to process effects from the outbox. It routes effects to the appropriate workflow handler based on the workflow_type.

Lifecycle

1. Create with [WorkflowRuntime::builder(store, WorkflowServiceConfig::default())]
2. Register workflows with WorkflowBuilder::register()
3. Configure with WorkflowBuilder::config()
4. Build with WorkflowBuilder::build_runtime()
5. Run with WorkflowRuntime::run() (not yet implemented)

Example

let runtime = WorkflowRuntime::builder(store, WorkflowServiceConfig::default())
    .register::<OrderWorkflow>(order_handler)
    .build_runtime()?;

// Run until shutdown signal
runtime.run(shutdown_signal).await?;

Implementations

impl<S> WorkflowRuntime<S>

where S: Store + WorkflowQueryStore,

pub fn builder( store: S, service_config: WorkflowServiceConfig, ) -> WorkflowBuilder<S>

Create a new runtime builder.

pub fn config(&self) -> &RuntimeConfig

Returns the runtime configuration.

pub fn worker_id(&self) -> &str

Returns the worker identifier.

pub fn workflow_count(&self) -> usize

Returns the number of registered workflows.

impl<S> WorkflowRuntime<S>

where S: Store + WorkflowQueryStore + OutboxStore,

pub async fn run<F>(self, shutdown: F) -> Result<()>

where F: Future<Output = ()> + Send,

Run the effect and timer workers until shutdown signal.

This method starts workers which poll the outbox:

• Effect workers: process immediate effects via handlers
• Timer workers: process due timers by routing embedded inputs

The number of workers is controlled by effect_workers and timer_workers in RuntimeConfig. Workers coordinate via FOR UPDATE SKIP LOCKED to avoid processing the same effect twice.

Shutdown Behavior

When the shutdown future completes:

1. All workers stop claiming new work
2. Wait for current work (if any) to complete
3. Return cleanly after timeout

Example

use tokio::signal;

let runtime = WorkflowRuntime::builder(pg_store, WorkflowServiceConfig::default())
    .register(order_handler)
    .config(RuntimeConfig {
        effect_workers: 4,  // 4 parallel effect workers
        ..Default::default()
    })
    .build_runtime()?;

// Run until Ctrl+C
runtime.run(async { signal::ctrl_c().await.ok(); }).await?;

pub async fn fetch_dead_letters( &self, query: DeadLetterQuery, ) -> Result<Vec<DeadLetter>>

Fetch dead-lettered effects.

Returns effects that have exceeded the configured max_attempts and are no longer being retried.

Example

use ironflow::runtime::outbox::DeadLetterQuery;

// Fetch all dead letters
let dead_letters = runtime.fetch_dead_letters(DeadLetterQuery::new()).await?;

// Fetch dead letters for a specific workflow type
let order_dead_letters = runtime
    .fetch_dead_letters(DeadLetterQuery::new().workflow_type("order"))
    .await?;

pub async fn count_dead_letters(&self, query: DeadLetterQuery) -> Result<u64>

Count dead-lettered effects.

Useful for monitoring and alerting on dead letter queue size.

Example

let count = runtime.count_dead_letters(DeadLetterQuery::new()).await?;
println!("Dead letters: {}", count);

pub async fn retry_dead_letter(&self, effect_id: Uuid) -> Result<bool>

Retry a dead-lettered effect.

Resets the effect’s attempt count to 0, making it available for processing again by the effect worker.

Returns Ok(true) if the effect was found and reset, Ok(false) if the effect was not found or already processed.

Example

let dead_letters = runtime.fetch_dead_letters(DeadLetterQuery::new()).await?;
for dl in dead_letters {
    runtime.retry_dead_letter(dl.id).await?;
}

pub async fn fetch_timer_dead_letters( &self, query: DeadLetterQuery, ) -> Result<Vec<DeadLetter>>

Fetch dead-lettered timers.

pub async fn count_timer_dead_letters( &self, query: DeadLetterQuery, ) -> Result<u64>

Count dead-lettered timers.

pub async fn retry_timer_dead_letter(&self, timer_id: Uuid) -> Result<bool>

Retry a dead-lettered timer.

Trait Implementations

impl<S> Clone for WorkflowRuntime<S>

where S: Store + WorkflowQueryStore + Clone,

fn clone(&self) -> WorkflowRuntime<S>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.