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 with sqlx pool defaults and run migrations.

Equivalent to Self::connect_with_options called with PoolOptions::default(). Use that method instead when you need to tune pool size or session-level timeouts.

Errors

Returns an error if the connection or migration fails.

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

Connect to Postgres with explicit pool options and run migrations.

Field-level details on each option are documented on PoolOptions. statement_timeout and idle_in_transaction_session_timeout are installed via an after_connect hook that runs SET on every freshly-acquired connection.

Errors

Returns an error if the connection, the SET hooks, or the migration fail.

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

Use an existing connection pool and run migrations.

Prefer Self::connect_with_options when you only want to tune standard pool knobs — this method is meant for callers who need full control over the sqlx PgPool (custom TLS, listeners, etc.).

Errors

Returns an error if the migration fails.

Trait Implementations

impl<C: Clone> Clone for PostgresBackend<C>

fn clone(&self) -> PostgresBackend<C>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<C> SignalStore for PostgresBackend<C>

where C: SnapshotCodec,

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<TaskId>, ) -> 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: SnapshotCodec,

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

Save a workflow snapshot.

async fn save_task_result( &self, instance_id: &str, task_id: &TaskId, 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: SnapshotCodec,

async fn claim_task( &self, instance_id: &str, task_id: &TaskId, 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: &TaskId, worker_id: &str, ) -> Result<(), BackendError>

Release a task claim.

async fn extend_task_claim( &self, instance_id: &str, task_id: &TaskId, 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, aging_interval: Duration, worker_tags: &[String], ) -> Result<Vec<AvailableTask>, BackendError>

Find available tasks across all workflow instances.

async fn wait_for_wakeup( &self, timeout: Duration, ) -> Result<Option<TaskWakeupHint>, BackendError>

Block until a wakeup arrives or timeout elapses. Some(hint) when a producer attached one; None for hint-less wakeups, timeout, or backends without a notification channel.

async fn find_hinted_task( &self, hint: &TaskWakeupHint, ) -> Result<Option<AvailableTask>, BackendError>

Resolve a wakeup hint into a claimable AvailableTask, or None if the snapshot moved on / is claim-blocked / signal-blocked. Does NOT claim — the caller runs the normal claim+execute pipeline. Default returns None; backends opt in for the targeted lookup.

impl<C> TaskResultStore for PostgresBackend<C>

where C: SnapshotCodec,

async fn load_task_result( &self, instance_id: &str, task_id: &TaskId, ) -> Result<Option<Bytes>, BackendError>

Load a single task result by task ID.