chronon_backend_postgres/
lib.rs

1//! `PostgreSQL` [`SchedulerStore`](chronon_core::store::SchedulerStore) for Chronon.
2//!
3//! **When to use:** shared durable storage for **embedded** or **coordinator–worker** clusters.
4//! For higher claim throughput, wrap with `PostgresRedisSchedulerStore`
5//! (`chronon-backend-redis`).
6//!
7//! Getting started:
8//! [Embedded](https://docs.rs/uf-chronon/latest/chronon/index.html#embedded-one-process) /
9//! [Coordinator–worker](https://docs.rs/uf-chronon/latest/chronon/index.html#coordinator-worker-split).
10//!
11//! ## Stack position
12//!
13//! ```text
14//! chronon (public crate, `postgres` feature) → chronon-backend-postgres → chronon-backend-sql-common → chronon-core
15//! ```
16//!
17//! ## Entry points
18//!
19//! - [`PostgresSchedulerStore::connect`] — open a pool and bootstrap schema
20//! - [`PostgresSchedulerStore::connect_isolated`] — isolated schema for parallel tests
21//! - [`postgres_test_url`] — resolve test URL from `CHRONON_POSTGRES_URL` / `CHRONON_TEST_POSTGRES_URL`
22//!
23//! ## Embedded
24//!
25//! ```ignore
26//! use std::sync::Arc;
27//! use chronon::prelude::*;
28//! use chronon::PostgresSchedulerStore;
29//!
30//! let url = std::env::var("CHRONON_POSTGRES_URL")?;
31//! let store = PostgresSchedulerStore::connect(&url).await?;
32//! let chronon = ChrononBuilder::new()
33//!     .scheduler_store(Arc::new(store))
34//!     .context_factory(Arc::new(JsonScriptContextFactory))
35//!     .embedded()
36//!     .auto_registry()
37//!     .build()?;
38//! ```
39//!
40//! Runnable: `cargo run -p uf-chronon --example postgres_boot --features postgres`
41//!
42//! ## Coordinator binary
43//!
44//! Shared `CHRONON_POSTGRES_URL` with workers. Tick only:
45//!
46//! ```ignore
47//! use std::sync::Arc;
48//! use chronon::prelude::*;
49//! use chronon::PostgresSchedulerStore;
50//!
51//! let url = std::env::var("CHRONON_POSTGRES_URL")?;
52//! let store = PostgresSchedulerStore::connect(&url).await?;
53//! let mut chronon = ChrononBuilder::new()
54//!     .scheduler_store(Arc::new(store))
55//!     .context_factory(Arc::new(JsonScriptContextFactory))
56//!     .instance_id("coordinator-0")
57//!     .coordinator_only()
58//!     .build()?;
59//! chronon.scheduler.init_partitions().await;
60//! chronon.run().await?;
61//! ```
62//!
63//! Runnable: `cargo run -p uf-chronon --example postgres_coordinator_daemon --features postgres`
64//!
65//! ## Worker binary
66//!
67//! Same Postgres URL, unique `CHRONON_INSTANCE_ID`, scripts via `.auto_registry()`:
68//!
69//! ```ignore
70//! use std::sync::Arc;
71//! use chronon::prelude::*;
72//! use chronon::PostgresSchedulerStore;
73//!
74//! let url = std::env::var("CHRONON_POSTGRES_URL")?;
75//! let store = PostgresSchedulerStore::connect(&url).await?;
76//! let mut chronon = ChrononBuilder::new()
77//!     .scheduler_store(Arc::new(store))
78//!     .context_factory(Arc::new(JsonScriptContextFactory))
79//!     .instance_id(std::env::var("CHRONON_INSTANCE_ID").unwrap_or_else(|_| "worker-1".into()))
80//!     .auto_registry()
81//!     .worker("general")
82//!     .build()?;
83//! chronon.run().await?;
84//! ```
85//!
86//! Runnable: `cargo run -p uf-chronon --example postgres_worker_daemon --features postgres`
87//!
88//! Production claim path: [Postgres + Redis](../chronon_backend_redis/index.html#coordinator-binary).
89
90mod bootstrap;
91
92use chronon_backend_sql_common::SqlSchedulerStore;
93use chronon_core::Result;
94use sqlx::PgPool;
95
96pub use bootstrap::{postgres_store_from_env, postgres_test_url};
97
98/// PostgreSQL-backed scheduler store.
99///
100/// Shared durable storage for coordinator–worker clusters (and embedded when you already run
101/// Postgres). Pass a connection URL to [`Self::connect`]; daemons often use
102/// `CHRONON_POSTGRES_URL` / [`postgres_test_url`].
103///
104/// Split examples: [coordinator](index.html#coordinator-binary) /
105/// [worker](index.html#worker-binary).
106///
107/// For higher claim throughput, wrap with `PostgresRedisSchedulerStore` from
108/// `chronon-backend-redis` (`postgres` + `redis` features). Enable the public crate `postgres`
109/// feature to re-export this type.
110///
111/// # Examples
112///
113/// ```rust,no_run
114/// use chronon_backend_postgres::PostgresSchedulerStore;
115///
116/// # async fn example() -> chronon_core::Result<()> {
117/// let store = PostgresSchedulerStore::connect(
118///     "postgres://user:pass@localhost/chronon",
119/// )
120/// .await?;
121/// # let _ = store;
122/// # Ok(())
123/// # }
124/// ```
125///
126/// Runnable: `cargo run -p uf-chronon --example postgres_boot --features postgres`.
127pub struct PostgresSchedulerStore {
128    inner: SqlSchedulerStore,
129}
130
131impl PostgresSchedulerStore {
132    /// Connect using a `PostgreSQL` connection URL.
133    ///
134    /// # Examples
135    ///
136    /// ```rust,no_run
137    /// use chronon_backend_postgres::PostgresSchedulerStore;
138    ///
139    /// # async fn example() -> chronon_core::Result<()> {
140    /// let store = PostgresSchedulerStore::connect(
141    ///     "postgres://user:pass@localhost/chronon",
142    /// )
143    /// .await?;
144    /// # Ok(())
145    /// # }
146    /// ```
147    ///
148    /// # Errors
149    ///
150    /// Returns a storage error when the pool cannot connect or schema bootstrap fails.
151    pub async fn connect(url: &str) -> Result<Self> {
152        let inner = SqlSchedulerStore::connect_postgres(url).await?;
153        Ok(Self { inner })
154    }
155
156    /// Connect with an isolated schema (for parallel tests).
157    ///
158    /// `schema` must be an allowlisted identifier (`^[A-Za-z_][A-Za-z0-9_]*$`, max 63 chars)
159    /// or connect fails with [`chronon_core::ChrononError::ParamError`] before any DDL.
160    ///
161    /// # Examples
162    ///
163    /// ```rust,no_run
164    /// use chronon_backend_postgres::PostgresSchedulerStore;
165    ///
166    /// # async fn example() -> chronon_core::Result<()> {
167    /// let store = PostgresSchedulerStore::connect_isolated(
168    ///     "postgres://user:pass@localhost/chronon",
169    ///     "chronon_test_schema",
170    /// )
171    /// .await?;
172    /// # Ok(())
173    /// # }
174    /// ```
175    ///
176    /// # Errors
177    ///
178    /// Returns a parameter error for invalid schema names, or a storage error when schema
179    /// creation, pool connect, or bootstrap fails.
180    pub async fn connect_isolated(url: &str, schema: &str) -> Result<Self> {
181        let inner = SqlSchedulerStore::connect_postgres_isolated(url, schema).await?;
182        Ok(Self { inner })
183    }
184
185    /// Attach to an existing isolated schema (no DDL bootstrap; for multi-process workers).
186    ///
187    /// # Errors
188    ///
189    /// Returns a storage error when the pool cannot be opened.
190    pub async fn attach_isolated(url: &str, schema: &str) -> Result<Self> {
191        let inner = SqlSchedulerStore::attach_postgres_isolated(url, schema).await?;
192        Ok(Self { inner })
193    }
194
195    /// Drop an isolated bench/test schema (multibench cell reset).
196    ///
197    /// # Errors
198    ///
199    /// Returns a storage error when the admin connection or DDL fails.
200    pub async fn drop_isolated_schema(url: &str, schema: &str) -> Result<()> {
201        SqlSchedulerStore::drop_postgres_schema(url, schema).await
202    }
203
204    /// Wrap an existing pool (schema bootstrap runs).
205    ///
206    /// # Examples
207    ///
208    /// ```rust,no_run
209    /// use chronon_backend_postgres::PostgresSchedulerStore;
210    /// use sqlx::postgres::PgPoolOptions;
211    ///
212    /// # async fn example() -> chronon_core::Result<()> {
213    /// let pool = PgPoolOptions::new()
214    ///     .connect("postgres://localhost/chronon")
215    ///     .await
216    ///     .expect("pool");
217    /// let store = PostgresSchedulerStore::from_pool(pool).await?;
218    /// # Ok(())
219    /// # }
220    /// ```
221    ///
222    /// # Errors
223    ///
224    /// Returns a storage error when schema bootstrap fails.
225    pub async fn from_pool(pool: PgPool) -> Result<Self> {
226        let inner = SqlSchedulerStore::from_postgres_pool(pool).await?;
227        Ok(Self { inner })
228    }
229
230    /// Underlying connection pool.
231    ///
232    /// # Panics
233    ///
234    /// Panics if the inner pool is not `PostgreSQL` (internal invariant violation).
235    #[must_use]
236    pub fn pool(&self) -> &PgPool {
237        match self.inner.pool() {
238            chronon_backend_sql_common::SqlPool::Postgres(pool) => pool,
239            chronon_backend_sql_common::SqlPool::Sqlite(_) => {
240                panic!("postgres backend has non-postgres pool")
241            }
242        }
243    }
244}
245
246impl std::fmt::Debug for PostgresSchedulerStore {
247    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248        f.debug_struct("PostgresSchedulerStore")
249            .finish_non_exhaustive()
250    }
251}
252
253chronon_backend_sql_common::delegate_scheduler_store!(PostgresSchedulerStore, inner);
chronon_backend_postgres/lib.rs

chronon_backend_postgres/
lib.rs
