boson_backend_postgres/
lib.rs

1//! `PostgreSQL` [`QueueBackend`](boson_core::QueueBackend) for Boson.
2//!
3//! **When to use:** shared durable state for Mode 1 or Mode 2 (enqueue hosts + worker binaries
4//! against the same database). Enable via the `boson` crate `postgres` feature.
5//!
6//! Mode 2 workers should set a unique `worker_id` and `lease_ttl_secs > 0`. See the
7//! [`boson`](https://docs.rs/uf-boson) crate
8//! [Mode 2](https://docs.rs/uf-boson/latest/boson/index.html#mode-2--remote-worker-two-binaries).
9//!
10//! ## Entry points
11//!
12//! - [`PostgresQueueBackend::connect`] — open a pool and bootstrap schema
13//! - [`install_default_postgres_backend`] — register on the global [`QueueRouter`](boson_core::QueueRouter)
14//!
15//! ## Mode 2 — Enqueue binary
16//!
17//! Shared `DATABASE_URL` with the worker. No claim loop in this process:
18//!
19//! ```rust,no_run
20//! use std::sync::Arc;
21//!
22//! use boson_backend_postgres::PostgresQueueBackend;
23//! use boson_core::JsonExecutionContextFactory;
24//! use boson_runtime::{configure, Boson};
25//!
26//! # async fn boot_enqueue() -> boson_core::Result<()> {
27//! let url = std::env::var("DATABASE_URL")
28//!     .unwrap_or_else(|_| "postgres://localhost/boson".into());
29//! let backend = PostgresQueueBackend::connect(&url).await?;
30//! let boson = Boson::builder()
31//!     .queue_backend(Arc::new(backend))
32//!     .execution_context_factory(JsonExecutionContextFactory)
33//!     .auto_registry()
34//!     .without_worker()
35//!     .build()?;
36//! configure(boson);
37//! // MyTask::send_with(...).await?;
38//! # Ok(())
39//! # }
40//! ```
41//!
42//! ## Mode 2 — Worker binary
43//!
44//! Same database URL, unique `worker_id`, and `lease_ttl_secs > 0`:
45//!
46//! ```rust,no_run
47//! use std::sync::Arc;
48//!
49//! use boson_backend_postgres::PostgresQueueBackend;
50//! use boson_core::JsonExecutionContextFactory;
51//! use boson_runtime::Boson;
52//!
53//! # async fn boot_worker() -> boson_core::Result<()> {
54//! let url = std::env::var("DATABASE_URL")
55//!     .unwrap_or_else(|_| "postgres://localhost/boson".into());
56//! let backend = PostgresQueueBackend::connect(&url).await?;
57//! let _boson = Boson::builder()
58//!     .queue_backend(Arc::new(backend))
59//!     .execution_context_factory(JsonExecutionContextFactory)
60//!     .worker_id(std::env::var("BOSON_WORKER_ID").unwrap_or_else(|_| "worker-1".into()))
61//!     .lease_ttl_secs(30)
62//!     .auto_registry()
63//!     .build()?;
64//! # Ok(())
65//! # }
66//! ```
67//!
68//! Other Mode 2 backends:
69//! [`SQLite`](../boson_backend_sqlite/index.html#mode-2--enqueue-binary),
70//! [Redis](../boson_backend_redis/index.html#mode-2--enqueue-binary),
71//! [NATS](../boson_backend_nats/index.html#mode-2--enqueue-binary).
72
73mod bootstrap;
74
75use boson_backend_sql_common::SqlQueueBackend;
76use boson_core::Result;
77use sqlx::PgPool;
78
79pub use bootstrap::{
80    install_default_postgres_backend, install_isolated_postgres_backend, postgres_test_url,
81};
82
83/// PostgreSQL-backed queue backend.
84///
85/// Mode 2 examples: [enqueue](index.html#mode-2--enqueue-binary) /
86/// [worker](index.html#mode-2--worker-binary).
87pub struct PostgresQueueBackend {
88    inner: SqlQueueBackend,
89}
90
91impl PostgresQueueBackend {
92    /// Connect to `PostgreSQL` at `url`.
93    ///
94    /// # Errors
95    ///
96    /// Returns an error when the pool cannot connect or schema bootstrap fails.
97    pub async fn new(url: &str) -> Result<Self> {
98        Self::connect(url).await
99    }
100
101    /// Connect using a `PostgreSQL` connection URL and wire into [`Boson`](https://docs.rs/boson-runtime).
102    ///
103    /// See crate-level [Mode 2 — Enqueue binary](index.html#mode-2--enqueue-binary) and
104    /// [Mode 2 — Worker binary](index.html#mode-2--worker-binary).
105    ///
106    /// # Examples
107    ///
108    /// ```rust,no_run
109    /// use std::sync::Arc;
110    ///
111    /// use boson_backend_postgres::PostgresQueueBackend;
112    /// use boson_core::JsonExecutionContextFactory;
113    /// use boson_runtime::Boson;
114    ///
115    /// # async fn connect() -> boson_core::Result<()> {
116    /// let url = std::env::var("DATABASE_URL")
117    ///     .unwrap_or_else(|_| "postgres://localhost/boson".into());
118    /// let backend = PostgresQueueBackend::connect(&url).await?;
119    /// let _boson = Boson::builder()
120    ///     .queue_backend(Arc::new(backend))
121    ///     .execution_context_factory(JsonExecutionContextFactory)
122    ///     .worker_id("worker-1")
123    ///     .lease_ttl_secs(30) // Mode 2 multi-process
124    ///     .auto_registry()
125    ///     .build()?;
126    /// # Ok(())
127    /// # }
128    /// ```
129    ///
130    /// # Errors
131    ///
132    /// Returns an error when the pool cannot connect or schema bootstrap fails.
133    pub async fn connect(url: &str) -> Result<Self> {
134        let inner = SqlQueueBackend::connect_postgres(url).await?;
135        Ok(Self { inner })
136    }
137
138    /// Connect with an isolated schema (for parallel tests).
139    ///
140    /// # Errors
141    ///
142    /// Returns an error when schema creation, pool connect, or bootstrap fails.
143    pub async fn connect_isolated(url: &str, schema: &str) -> Result<Self> {
144        let inner = SqlQueueBackend::connect_postgres_isolated(url, schema).await?;
145        Ok(Self { inner })
146    }
147
148    /// Wrap an existing pool (schema bootstrap runs).
149    ///
150    /// # Errors
151    ///
152    /// Returns an error when schema bootstrap fails.
153    pub async fn from_pool(pool: PgPool) -> Result<Self> {
154        let inner = SqlQueueBackend::from_postgres_pool(pool).await?;
155        Ok(Self { inner })
156    }
157
158    /// Underlying connection pool.
159    ///
160    /// # Panics
161    ///
162    /// Panics if the inner pool is not `PostgreSQL` (internal invariant violation).
163    #[must_use]
164    pub fn pool(&self) -> &PgPool {
165        match self.inner.pool() {
166            boson_backend_sql_common::SqlPool::Postgres(pool) => pool,
167            boson_backend_sql_common::SqlPool::Sqlite(_) => {
168                panic!("postgres backend has non-postgres pool")
169            }
170        }
171    }
172}
173
174impl std::fmt::Debug for PostgresQueueBackend {
175    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
176        f.debug_struct("PostgresQueueBackend").finish_non_exhaustive()
177    }
178}
179
180boson_backend_sql_common::delegate_queue_backend!(PostgresQueueBackend, inner);
boson_backend_postgres/lib.rs

boson_backend_postgres/
lib.rs
