Struct PostgresProvider 

pub struct PostgresProvider { /* private fields */ }

PostgreSQL-based provider for Duroxide durable orchestrations.

Implements the Provider and ProviderAdmin traits from Duroxide, storing orchestration state, history, and work queues in PostgreSQL.

Example

use duroxide_pg::PostgresProvider;

// Connect using DATABASE_URL or explicit connection string
let provider = PostgresProvider::new("postgres://localhost/mydb").await?;

// Or use a custom schema for isolation
let provider = PostgresProvider::new_with_schema(
    "postgres://localhost/mydb",
    Some("my_app"),
).await?;

Implementations

impl PostgresProvider

pub async fn new(database_url: &str) -> Result<Self>

pub async fn new_with_schema( database_url: &str, schema_name: Option<&str>, ) -> Result<Self>

pub async fn initialize_schema(&self) -> Result<()>

pub fn pool(&self) -> &PgPool

Get the database pool (for testing)

pub fn schema_name(&self) -> &str

Get the schema name (for testing)

pub async fn cleanup_schema(&self) -> Result<()>

Clean up schema after tests (drops all tables and optionally the schema)

SAFETY: Never drops the “public” schema itself, only tables within it. Only drops the schema if it’s a custom schema (not “public”).

Trait Implementations

impl Provider for PostgresProvider

fn fetch_orchestration_item<'life0, 'async_trait>( &'life0 self, lock_timeout: Duration, _poll_timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<Option<OrchestrationItem>, ProviderError>> + Send + 'async_trait>>

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

Fetch the next orchestration work item atomically.

fn ack_orchestration_item<'life0, 'life1, 'async_trait>( &'life0 self, lock_token: &'life1 str, execution_id: u64, history_delta: Vec<Event>, worker_items: Vec<WorkItem>, orchestrator_items: Vec<WorkItem>, metadata: ExecutionMetadata, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Acknowledge successful orchestration processing atomically.

fn abandon_orchestration_item<'life0, 'life1, 'async_trait>( &'life0 self, lock_token: &'life1 str, delay: Option<Duration>, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Abandon orchestration processing (used for errors/retries).

fn read<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, ProviderError>> + Send + 'async_trait>>

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

Read the full history for the latest execution of an instance.

fn append_with_execution<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, execution_id: u64, new_events: Vec<Event>, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Append events to a specific execution.

fn enqueue_for_worker<'life0, 'async_trait>( &'life0 self, item: WorkItem, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Enqueue an activity execution request.

fn fetch_work_item<'life0, 'async_trait>( &'life0 self, lock_timeout: Duration, _poll_timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<Option<(WorkItem, String)>, ProviderError>> + Send + 'async_trait>>

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

Dequeue a single work item with peek-lock semantics.

fn ack_work_item<'life0, 'life1, 'async_trait>( &'life0 self, token: &'life1 str, completion: WorkItem, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Acknowledge successful processing of a work item.

fn renew_work_item_lock<'life0, 'life1, 'async_trait>( &'life0 self, token: &'life1 str, extend_for: Duration, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Renew the lock on a worker queue item.

fn enqueue_for_orchestrator<'life0, 'async_trait>( &'life0 self, item: WorkItem, delay: Option<Duration>, ) -> Pin<Box<dyn Future<Output = Result<(), ProviderError>> + Send + 'async_trait>>

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

Enqueue a work item to the orchestrator queue.

fn read_with_execution<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, execution_id: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, ProviderError>> + Send + 'async_trait>>

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

Read the full event history for a specific execution within an instance.

fn as_management_capability(&self) -> Option<&dyn ProviderAdmin>

Check if this provider implements management capabilities.

impl ProviderAdmin for PostgresProvider

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

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

List all known instance IDs.

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

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

List instances matching a status filter.

fn list_executions<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<u64>, ProviderError>> + Send + 'async_trait>>

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

List all execution IDs for an instance.

fn read_history_with_execution_id<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, execution_id: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, ProviderError>> + Send + 'async_trait>>

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

Read the full event history for a specific execution within an instance.

fn read_history<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, ProviderError>> + Send + 'async_trait>>

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

Read the full event history for the latest execution of an instance.

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

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

Get the latest (current) execution ID for an instance.

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

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

Get comprehensive information about an instance.

fn get_execution_info<'life0, 'life1, 'async_trait>( &'life0 self, instance: &'life1 str, execution_id: u64, ) -> Pin<Box<dyn Future<Output = Result<ExecutionInfo, ProviderError>> + Send + 'async_trait>>

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

Get detailed information about a specific execution.

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

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

Get system-wide metrics for the orchestration engine.

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

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

Get the current depths of the internal work queues.