Struct PostgresQueueBackend 

pub struct PostgresQueueBackend { /* private fields */ }

PostgreSQL-backed queue backend.

Mode 2 examples: enqueue / worker.

Implementations

impl PostgresQueueBackend

pub async fn new(url: &str) -> Result<PostgresQueueBackend, BosonError>

Connect to PostgreSQL at url.

Errors

Returns an error when the pool cannot connect or schema bootstrap fails.

pub async fn connect(url: &str) -> Result<PostgresQueueBackend, BosonError>

Connect using a PostgreSQL connection URL and wire into Boson.

See crate-level Mode 2 — Enqueue binary and Mode 2 — Worker binary.

Examples

use std::sync::Arc;

use boson_backend_postgres::PostgresQueueBackend;
use boson_core::JsonExecutionContextFactory;
use boson_runtime::Boson;

let url = std::env::var("DATABASE_URL")
    .unwrap_or_else(|_| "postgres://localhost/boson".into());
let backend = PostgresQueueBackend::connect(&url).await?;
let _boson = Boson::builder()
    .queue_backend(Arc::new(backend))
    .execution_context_factory(JsonExecutionContextFactory)
    .worker_id("worker-1")
    .lease_ttl_secs(30) // Mode 2 multi-process
    .auto_registry()
    .build()?;

Errors

Returns an error when the pool cannot connect or schema bootstrap fails.

pub async fn connect_isolated( url: &str, schema: &str, ) -> Result<PostgresQueueBackend, BosonError>

Connect with an isolated schema (for parallel tests).

Errors

Returns an error when schema creation, pool connect, or bootstrap fails.

pub async fn from_pool( pool: Pool<Postgres>, ) -> Result<PostgresQueueBackend, BosonError>

Wrap an existing pool (schema bootstrap runs).

Errors

Returns an error when schema bootstrap fails.

pub fn pool(&self) -> &Pool<Postgres>

Underlying connection pool.

Panics

Panics if the inner pool is not PostgreSQL (internal invariant violation).

Trait Implementations

impl Debug for PostgresQueueBackend

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl QueueBackend for PostgresQueueBackend

fn upsert_job<'life0, 'life1, 'async_trait>( &'life0 self, job: &'life1 Job, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Persist or update a job row.

fn enqueue_with_policies<'life0, 'life1, 'async_trait>( &'life0 self, job: Job, task_config: &'life1 TaskConfig, ) -> Pin<Box<dyn Future<Output = Result<(String, JobEnqueueDisposition), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Insert job with idempotency semantics (see trait-level contract).

fn get_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Load one job by id.

fn list_jobs<'life0, 'async_trait>( &'life0 self, status_filter: Option<JobStatus>, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresQueueBackend: 'async_trait,

List jobs with optional status filter and pagination.

fn cancel_job_if_active<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Cancel a job if it is still active (queued or running).

fn try_claim_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Atomically claim a queued job for execution.

fn revert_job_to_queued<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Revert a job to queued (retry path).

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

where 'life0: 'async_trait, PostgresQueueBackend: 'async_trait,

Distinct pool names with queued jobs.

fn list_queued_for_pool_sorted<'life0, 'life1, 'async_trait>( &'life0 self, pool: &'life1 str, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Queued jobs for one pool, sorted by priority then created time.

fn count_jobs<'life0, 'async_trait>( &'life0 self, status_filter: Option<JobStatus>, ) -> Pin<Box<dyn Future<Output = Result<u64, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresQueueBackend: 'async_trait,

Count jobs, optionally filtered by status.

fn count_jobs_for_task<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, status: Option<JobStatus>, ) -> Pin<Box<dyn Future<Output = Result<u64, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Count jobs for one task, optionally filtered by status.

fn count_active_jobs_for_task<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u32, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Count active (queued + running) jobs for rate-limit checks.

fn find_nonterminal_by_idempotency_key<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Find non-terminal job by idempotency key.

fn upsert_run<'life0, 'life1, 'async_trait>( &'life0 self, run: &'life1 Run, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Persist or update a run row.

fn get_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Run>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Load one run by id.

fn list_runs<'life0, 'life1, 'async_trait>( &'life0 self, job_id_filter: Option<&'life1 str>, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

List runs with optional job filter and pagination.

fn finish_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, status: RunStatus, duration_ms: Option<i64>, error_message: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Mark a run terminal with outcome fields.

fn count_runs<'life0, 'life1, 'async_trait>( &'life0 self, job_id_filter: Option<&'life1 str>, ) -> Pin<Box<dyn Future<Output = Result<u64, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Count runs, optionally filtered by job id.

fn count_runs_since<'life0, 'async_trait>( &'life0 self, since: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<u64, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresQueueBackend: 'async_trait,

Count runs with started_at >= since.

fn task_run_stats<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<TaskRunStats, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Aggregate run totals for one task.

fn get_task_config<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskConfig>, BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Load task config by name.

fn upsert_task_config<'life0, 'life1, 'async_trait>( &'life0 self, config: &'life1 TaskConfig, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Persist task config.

fn try_claim_run_lease<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: &'life1 str, worker_id: &'life2 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, BosonError>> + Send + 'async_trait>>

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

Attempt to claim a run lease for job_id (stored as lease task id).

fn extend_lease<'life0, 'life1, 'async_trait>( &'life0 self, lease_id: &'life1 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Extend lease TTL for a held lease.

fn release_lease<'life0, 'life1, 'async_trait>( &'life0 self, lease_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), BosonError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresQueueBackend: 'async_trait,

Release a held lease.

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

where 'life0: 'async_trait, PostgresQueueBackend: 'async_trait,

Expired leases as (lease_record_id, job_id).

fn pop_claim_from_pool<'life0, 'life1, 'async_trait>( &'life0 self, _pool: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>, BosonError>> + Send + 'async_trait>>

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

Atomically claim the highest-priority queued job from pool when supported.