chronon_backend_redis/composite/
mod.rs

1//! PostgreSQL (or any SQL store) plus Redis ready-queue composite.
2//!
3//! Internal — used by [`PostgresRedisSchedulerStore`](crate::PostgresRedisSchedulerStore).
4
5mod claim;
6
7use std::fmt;
8use std::sync::Arc;
9
10use async_trait::async_trait;
11use chrono::{DateTime, Utc};
12
13use chronon_backend_sql_common::run_pool_key;
14use chronon_core::models::{Run, RunStatus};
15use chronon_core::store::SchedulerStore;
16use chronon_core::Result;
17
18use crate::queue::RedisQueueLayer;
19
20/// SQL persistence with Redis-backed run claim ordering.
21///
22/// Production Mode 2 store: job/run rows stay in Postgres (or any [`SchedulerStore`]); workers
23/// claim through Redis ZSETs for higher throughput. Requires facade features `postgres` and
24/// `redis`. Build with [`Self::new`] after connecting both layers.
25///
26/// Mode 2 examples: [coordinator](index.html#mode-2--coordinator-binary) /
27/// [worker](index.html#mode-2--worker-binary).
28///
29/// See [`crate::RedisQueueLayer`] for URL / prefix / cluster env vars
30/// (`CHRONON_REDIS_URL`, `CHRONON_REDIS_CLUSTER_URLS`, `CHRONON_REDIS_HASH_TAGS`).
31///
32/// Runnable: `cargo run -p uf-chronon --example postgres_redis_boot --features postgres,redis`
33/// and the `coordinator_daemon` / `worker_daemon` pair.
34pub struct PostgresRedisSchedulerStore {
35    sql: Arc<dyn SchedulerStore>,
36    redis: RedisQueueLayer,
37}
38
39impl std::fmt::Debug for PostgresRedisSchedulerStore {
40    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::fmt::Result {
41        f.debug_struct("PostgresRedisSchedulerStore")
42            .finish_non_exhaustive()
43    }
44}
45
46impl PostgresRedisSchedulerStore {
47    /// Wrap a SQL store and Redis queue layer.
48    ///
49    /// `create_run` writes SQL then enqueues Redis for queued runs; `claim_next_queued` pops
50    /// Redis then updates SQL lease state.
51    ///
52    /// # Examples
53    ///
54    /// ```rust,no_run
55    /// use std::sync::Arc;
56    ///
57    /// use chronon_backend_postgres::PostgresSchedulerStore;
58    /// use chronon_backend_redis::{PostgresRedisSchedulerStore, RedisQueueLayer};
59    /// use chronon_core::store::SchedulerStore;
60    ///
61    /// # async fn example() -> chronon_core::Result<()> {
62    /// let sql: Arc<dyn SchedulerStore> = Arc::new(
63    ///     PostgresSchedulerStore::connect("postgres://localhost/chronon").await?,
64    /// );
65    /// let redis = RedisQueueLayer::connect("redis://127.0.0.1:6379", Some("myapp")).await?;
66    /// let store = PostgresRedisSchedulerStore::new(sql, redis);
67    /// # Ok(())
68    /// # }
69    /// ```
70    #[must_use]
71    pub fn new(sql: Arc<dyn SchedulerStore>, redis: RedisQueueLayer) -> Self {
72        Self { sql, redis }
73    }
74}
75
76#[async_trait]
77impl SchedulerStore for PostgresRedisSchedulerStore {
78    async fn upsert_job(
79        &self,
80        job: &chronon_core::models::Job,
81    ) -> Result<()> {
82        self.sql.upsert_job(job).await
83    }
84
85    async fn get_job(&self, job_id: &str) -> Result<Option<chronon_core::models::Job>> {
86        self.sql.get_job(job_id).await
87    }
88
89    async fn get_job_by_name(
90        &self,
91        job_name: &str,
92    ) -> Result<Option<chronon_core::models::Job>> {
93        self.sql.get_job_by_name(job_name).await
94    }
95
96    async fn list_jobs(&self) -> Result<Vec<chronon_core::models::Job>> {
97        self.sql.list_jobs().await
98    }
99
100    async fn list_due_jobs(
101        &self,
102        before: DateTime<Utc>,
103    ) -> Result<Vec<chronon_core::models::Job>> {
104        self.sql.list_due_jobs(before).await
105    }
106
107    async fn pause_job(&self, job_id: &str) -> Result<()> {
108        self.sql.pause_job(job_id).await
109    }
110
111    async fn resume_job(&self, job_id: &str) -> Result<()> {
112        self.sql.resume_job(job_id).await
113    }
114
115    async fn create_run(&self, run: &Run) -> Result<()> {
116        self.sql.create_run(run).await?;
117        if run.status == RunStatus::Queued {
118            let pool = run_pool_key(run.pool_id.as_deref());
119            self.redis
120                .enqueue_run(pool, &run.run_id, run.scheduled_for)
121                .await?;
122        }
123        Ok(())
124    }
125
126    async fn update_run(&self, run: &Run) -> Result<()> {
127        self.sql.update_run(run).await
128    }
129
130    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
131        self.sql.get_run(run_id).await
132    }
133
134    async fn list_runs_for_job(&self, job_id: &str, limit: usize) -> Result<Vec<Run>> {
135        self.sql.list_runs_for_job(job_id, limit).await
136    }
137
138    async fn list_runs_filtered(
139        &self,
140        job_id: Option<&str>,
141        status: Option<RunStatus>,
142        offset: usize,
143        limit: usize,
144    ) -> Result<Vec<Run>> {
145        self.sql
146            .list_runs_filtered(job_id, status, offset, limit)
147            .await
148    }
149
150    async fn claim_next_queued(
151        &self,
152        pool_id: &str,
153        worker_id: &str,
154        now: DateTime<Utc>,
155        lease_ttl_secs: i64,
156    ) -> Result<Option<Run>> {
157        claim::claim_next_queued(
158            &self.sql,
159            &self.redis,
160            pool_id,
161            worker_id,
162            now,
163            lease_ttl_secs,
164        )
165        .await
166    }
167
168    async fn claim_run_by_id(
169        &self,
170        run_id: &str,
171        pool_id: &str,
172        worker_id: &str,
173        now: DateTime<Utc>,
174        lease_ttl_secs: i64,
175    ) -> Result<Option<Run>> {
176        self.sql
177            .claim_run_by_id(run_id, pool_id, worker_id, now, lease_ttl_secs)
178            .await
179    }
180
181    async fn claim_runs_by_ids(
182        &self,
183        run_ids: &[&str],
184        pool_id: &str,
185        worker_id: &str,
186        now: DateTime<Utc>,
187        lease_ttl_secs: i64,
188    ) -> Result<Vec<Run>> {
189        self.sql
190            .claim_runs_by_ids(run_ids, pool_id, worker_id, now, lease_ttl_secs)
191            .await
192    }
193
194    async fn renew_run_lease(
195        &self,
196        run_id: &str,
197        worker_id: &str,
198        now: DateTime<Utc>,
199        lease_ttl_secs: i64,
200    ) -> Result<bool> {
201        self.sql
202            .renew_run_lease(run_id, worker_id, now, lease_ttl_secs)
203            .await
204    }
205
206    async fn append_revision(
207        &self,
208        revision: &chronon_core::models::JobRevision,
209    ) -> Result<()> {
210        self.sql.append_revision(revision).await
211    }
212
213    async fn list_revisions(
214        &self,
215        job_id: &str,
216    ) -> Result<Vec<chronon_core::models::JobRevision>> {
217        self.sql.list_revisions(job_id).await
218    }
219
220    async fn upsert_script(&self, script: &chronon_core::models::Script) -> Result<()> {
221        self.sql.upsert_script(script).await
222    }
223
224    async fn get_script(
225        &self,
226        script_name: &str,
227    ) -> Result<Option<chronon_core::models::Script>> {
228        self.sql.get_script(script_name).await
229    }
230
231    async fn try_claim_run_once(
232        &self,
233        job_id: &str,
234        claimed_by: &str,
235        now: DateTime<Utc>,
236        claim_ttl_secs: i64,
237    ) -> Result<bool> {
238        self.sql
239            .try_claim_run_once(job_id, claimed_by, now, claim_ttl_secs)
240            .await
241    }
242
243    async fn mark_run_once_completed(
244        &self,
245        job_id: &str,
246        completed_at: DateTime<Utc>,
247    ) -> Result<()> {
248        self.sql
249            .mark_run_once_completed(job_id, completed_at)
250            .await
251    }
252
253    async fn release_run_once_claim(
254        &self,
255        job_id: &str,
256        claimed_by: &str,
257        now: DateTime<Utc>,
258    ) -> Result<()> {
259        self.sql
260            .release_run_once_claim(job_id, claimed_by, now)
261            .await
262    }
263
264    async fn find_due_job_ids_in_partitions(
265        &self,
266        owned_partitions: &[u32],
267        due_until: DateTime<Utc>,
268        limit: u32,
269    ) -> Result<Vec<String>> {
270        self.sql
271            .find_due_job_ids_in_partitions(owned_partitions, due_until, limit)
272            .await
273    }
274
275    async fn min_next_run_at_in_partitions(
276        &self,
277        owned_partitions: &[u32],
278    ) -> Result<Option<DateTime<Utc>>> {
279        self.sql
280            .min_next_run_at_in_partitions(owned_partitions)
281            .await
282    }
283
284    async fn claim_job_for_tick(
285        &self,
286        job_id: &str,
287        claim_id: &str,
288        now: DateTime<Utc>,
289        lease_ttl_secs: i64,
290    ) -> Result<bool> {
291        self.sql
292            .claim_job_for_tick(job_id, claim_id, now, lease_ttl_secs)
293            .await
294    }
295
296    async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
297        self.sql.release_job_tick_claim(job_id).await
298    }
299
300    async fn persist_post_tick_job_state(
301        &self,
302        job_id: &str,
303        next_run_at: Option<DateTime<Utc>>,
304    ) -> Result<()> {
305        self.sql
306            .persist_post_tick_job_state(job_id, next_run_at)
307            .await
308    }
309
310    async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
311        self.sql.try_acquire_leader(instance_id, ttl_secs).await
312    }
313
314    async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
315        self.sql.renew_leader_lease(instance_id, ttl_secs).await
316    }
317
318    async fn get_leader(&self) -> Result<Option<chronon_core::models::SchedulerLeader>> {
319        self.sql.get_leader().await
320    }
321
322    async fn upsert_partition_assignment(
323        &self,
324        assignment: &chronon_core::models::PartitionAssignment,
325    ) -> Result<()> {
326        self.sql.upsert_partition_assignment(assignment).await
327    }
328
329    async fn list_partition_assignments(
330        &self,
331    ) -> Result<Vec<chronon_core::models::PartitionAssignment>> {
332        self.sql.list_partition_assignments().await
333    }
334
335    async fn register_worker(&self, worker: &chronon_core::models::Worker) -> Result<()> {
336        self.sql.register_worker(worker).await
337    }
338
339    async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
340        self.sql.heartbeat_worker(worker_id, at).await
341    }
342}
chronon_backend_redis/composite/mod.rs

chronon_backend_redis/composite/
mod.rs
