Struct PostgresBackend 

pub struct PostgresBackend<C> { /* private fields */ }

PostgreSQL persistence backend for Sayiir workflows.

Generic over a Codec that determines how snapshots are serialized into the BYTEA column. Use JsonCodec for human-readable storage with Postgres-side queryability, or a binary codec for faster (de)serialization.

Example (with sayiir-runtime JSON codec)

use sayiir_postgres::PostgresBackend;
use sayiir_runtime::serialization::JsonCodec;

let backend = PostgresBackend::<JsonCodec>::connect("postgresql://localhost/sayiir").await?;

Implementations

impl<C> PostgresBackend<C>

where C: Default,

pub async fn connect(url: &str) -> Result<Self, BackendError>

Connect to Postgres and run migrations.

Errors

Returns an error if the connection or migration fails.

pub async fn connect_with(pool: PgPool) -> Result<Self, BackendError>

Use an existing connection pool and run migrations.

Errors

Returns an error if the migration fails.

Trait Implementations

impl<C> SignalStore for PostgresBackend<C>

where C: Encoder + Decoder + EncodeValue<WorkflowSnapshot> + DecodeValue<WorkflowSnapshot>,

async fn store_signal( &self, instance_id: &str, kind: SignalKind, request: SignalRequest, ) -> Result<(), BackendError>

Store a signal (cancel or pause) for a workflow instance.

async fn get_signal( &self, instance_id: &str, kind: SignalKind, ) -> Result<Option<SignalRequest>, BackendError>

Get the pending signal of the given kind, if any.

async fn clear_signal( &self, instance_id: &str, kind: SignalKind, ) -> Result<(), BackendError>

Clear the signal of the given kind.

async fn send_event( &self, instance_id: &str, signal_name: &str, payload: Bytes, ) -> Result<(), BackendError>

Send an external event to a workflow instance.

async fn consume_event( &self, instance_id: &str, signal_name: &str, ) -> Result<Option<Bytes>, BackendError>

Consume the oldest buffered event for the given signal name, if any.

async fn check_and_cancel( &self, instance_id: &str, interrupted_at_task: Option<&str>, ) -> Result<bool, BackendError>

Atomically check for cancellation and transition to cancelled state.

async fn check_and_pause(&self, instance_id: &str) -> Result<bool, BackendError>

Atomically check for a pause request and transition to paused state.

async fn unpause( &self, instance_id: &str, ) -> Result<WorkflowSnapshot, BackendError>

Transition a paused workflow back to in-progress and return the updated snapshot.

impl<C> SnapshotStore for PostgresBackend<C>

where C: Encoder + Decoder + EncodeValue<WorkflowSnapshot> + DecodeValue<WorkflowSnapshot>,

async fn save_snapshot( &self, snapshot: &WorkflowSnapshot, ) -> Result<(), BackendError>

Save a workflow snapshot.

async fn save_task_result( &self, instance_id: &str, task_id: &str, output: Bytes, ) -> Result<(), BackendError>

Save a single task result atomically.

async fn load_snapshot( &self, instance_id: &str, ) -> Result<WorkflowSnapshot, BackendError>

Load a workflow snapshot by instance ID.

async fn delete_snapshot(&self, instance_id: &str) -> Result<(), BackendError>

Delete a workflow snapshot.

async fn list_snapshots(&self) -> Result<Vec<String>, BackendError>

List all snapshot instance IDs.

impl<C> TaskClaimStore for PostgresBackend<C>

where C: Encoder + Decoder + EncodeValue<WorkflowSnapshot> + DecodeValue<WorkflowSnapshot>,

async fn claim_task( &self, instance_id: &str, task_id: &str, worker_id: &str, ttl: Option<Duration>, ) -> Result<Option<TaskClaim>, BackendError>

Claim a task for execution by a worker node.

async fn release_task_claim( &self, instance_id: &str, task_id: &str, worker_id: &str, ) -> Result<(), BackendError>

Release a task claim.

async fn extend_task_claim( &self, instance_id: &str, task_id: &str, worker_id: &str, additional_duration: Duration, ) -> Result<(), BackendError>

Extend a task claim’s expiration time.

async fn find_available_tasks( &self, worker_id: &str, limit: usize, ) -> Result<Vec<AvailableTask>, BackendError>

Find available tasks across all workflow instances.