Struct PostgresBackend 

pub struct PostgresBackend { /* private fields */ }

PostgreSQL-based storage backend for RunnerQ.

This backend provides:

• Permanent persistence - No TTL, records are kept forever
• Multi-node concurrency - Safe dequeue with FOR UPDATE SKIP LOCKED
• Full ACID guarantees - Transactional consistency
• Cross-process events - Uses PostgreSQL LISTEN/NOTIFY for real-time events

Implementations

impl PostgresBackend

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

Create a new PostgreSQL backend.

Arguments

• database_url - PostgreSQL connection URL (e.g., postgres://user:pass@host:5432/db)
• queue_name - Name of the queue (used for namespacing). Must be a valid identifier: starts with letter or underscore, contains only letters, digits, underscores.

Example

let backend = PostgresBackend::new(
    "postgres://postgres:password@localhost:5432/runnerq",
    "my_queue"
).await?;

pub async fn with_config( database_url: &str, queue_name: &str, default_lease_ms: i64, pool_size: u32, ) -> Result<Self, StorageError>

Create a new PostgreSQL backend with custom configuration.

Arguments

• database_url - PostgreSQL connection URL
• queue_name - Name of the queue. Must be a valid identifier: starts with letter or underscore, contains only letters, digits, underscores.
• default_lease_ms - Default lease duration in milliseconds
• pool_size - Maximum number of connections in the pool

Trait Implementations

impl Clone for PostgresBackend

fn clone(&self) -> PostgresBackend

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl InspectionStorage for PostgresBackend

fn stats<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<QueueStats, StorageError>> + Send + 'async_trait>>

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

Get current queue statistics.

fn list_pending<'life0, 'async_trait>( &'life0 self, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ActivitySnapshot>, StorageError>> + Send + 'async_trait>>

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

List activities in the pending queue.

fn list_processing<'life0, 'async_trait>( &'life0 self, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ActivitySnapshot>, StorageError>> + Send + 'async_trait>>

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

List activities currently being processed.

fn list_scheduled<'life0, 'async_trait>( &'life0 self, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ActivitySnapshot>, StorageError>> + Send + 'async_trait>>

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

List activities scheduled for future execution.

fn list_completed<'life0, 'async_trait>( &'life0 self, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ActivitySnapshot>, StorageError>> + Send + 'async_trait>>

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

List completed activities (recent).

fn list_dead_letter<'life0, 'async_trait>( &'life0 self, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<DeadLetterRecord>, StorageError>> + Send + 'async_trait>>

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

List activities in the dead letter queue.

fn get_activity<'life0, 'async_trait>( &'life0 self, activity_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<Option<ActivitySnapshot>, StorageError>> + Send + 'async_trait>>

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

Get a specific activity by ID.

fn get_activity_events<'life0, 'async_trait>( &'life0 self, activity_id: Uuid, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ActivityEvent>, StorageError>> + Send + 'async_trait>>

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

Get recent events for an activity.

fn event_stream( &self, ) -> BoxStream<'static, Result<ActivityEvent, StorageError>>

Stream real-time activity events.

impl QueueStorage for PostgresBackend

fn enqueue<'life0, 'async_trait>( &'life0 self, activity: QueuedActivity, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + 'async_trait>>

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

Enqueue an activity for immediate or scheduled execution.

fn dequeue<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, worker_id: &'life1 str, _timeout: Duration, activity_types: Option<&'life2 [String]>, ) -> Pin<Box<dyn Future<Output = Result<Option<QueuedActivity>, StorageError>> + Send + 'async_trait>>

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

Try to dequeue one activity for processing.

fn ack_success<'life0, 'life1, 'async_trait>( &'life0 self, activity_id: Uuid, result: Option<Value>, worker_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + 'async_trait>>

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

Mark a dequeued activity as successfully completed.

fn ack_failure<'life0, 'life1, 'async_trait>( &'life0 self, activity_id: Uuid, failure: FailureKind, worker_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool, StorageError>> + Send + 'async_trait>>

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

Mark a dequeued activity as failed.

fn process_scheduled<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<u64, StorageError>> + Send + 'async_trait>>

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

Process scheduled activities that are ready to run.

fn schedules_natively(&self) -> bool

Whether this backend handles scheduled activities natively in dequeue().

fn requeue_expired<'life0, 'async_trait>( &'life0 self, batch_size: usize, ) -> Pin<Box<dyn Future<Output = Result<u64, StorageError>> + Send + 'async_trait>>

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

Requeue expired leases back to the ready queue.

fn extend_lease<'life0, 'async_trait>( &'life0 self, activity_id: Uuid, extend_by: Duration, ) -> Pin<Box<dyn Future<Output = Result<bool, StorageError>> + Send + 'async_trait>>

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

Extend the lease for an activity currently being processed.

fn store_result<'life0, 'async_trait>( &'life0 self, activity_id: Uuid, result: ActivityResult, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + 'async_trait>>

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

Store the result of a completed activity.

fn check_idempotency<'life0, 'life1, 'async_trait>( &'life0 self, activity: &'life1 QueuedActivity, ) -> Pin<Box<dyn Future<Output = Result<Option<Uuid>, StorageError>> + Send + 'async_trait>>

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

Evaluate idempotency rules before enqueueing.

impl ResultStorage for PostgresBackend

fn get_result<'life0, 'async_trait>( &'life0 self, activity_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<Option<ActivityResult>, StorageError>> + Send + 'async_trait>>

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

Retrieve a stored activity result.