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 coordinator–worker store: job/run rows stay in Postgres (or any [`SchedulerStore`]);
23/// workers claim through Redis ZSETs for higher throughput. Requires public crate features
24/// `postgres` and `redis`. Build with [`Self::new`] after connecting both layers.
25///
26/// Split examples: [coordinator](index.html#coordinator-binary) /
27/// [worker](index.html#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(&self, job: &chronon_core::models::Job) -> Result<()> {
79        self.sql.upsert_job(job).await
80    }
81
82    async fn get_job(&self, job_id: &str) -> Result<Option<chronon_core::models::Job>> {
83        self.sql.get_job(job_id).await
84    }
85
86    async fn get_job_by_name(&self, job_name: &str) -> Result<Option<chronon_core::models::Job>> {
87        self.sql.get_job_by_name(job_name).await
88    }
89
90    async fn list_jobs(&self) -> Result<Vec<chronon_core::models::Job>> {
91        self.sql.list_jobs().await
92    }
93
94    async fn list_due_jobs(&self, before: DateTime<Utc>) -> Result<Vec<chronon_core::models::Job>> {
95        self.sql.list_due_jobs(before).await
96    }
97
98    async fn pause_job(&self, job_id: &str) -> Result<()> {
99        self.sql.pause_job(job_id).await
100    }
101
102    async fn resume_job(&self, job_id: &str) -> Result<()> {
103        self.sql.resume_job(job_id).await
104    }
105
106    async fn create_run(&self, run: &Run) -> Result<()> {
107        self.sql.create_run(run).await?;
108        if run.status == RunStatus::Queued {
109            let pool = run_pool_key(run.pool_id.as_deref());
110            self.redis
111                .enqueue_run(pool, &run.run_id, run.scheduled_for)
112                .await?;
113        }
114        Ok(())
115    }
116
117    async fn update_run(&self, run: &Run) -> Result<()> {
118        self.sql.update_run(run).await
119    }
120
121    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
122        self.sql.get_run(run_id).await
123    }
124
125    async fn list_runs_for_job(&self, job_id: &str, limit: usize) -> Result<Vec<Run>> {
126        self.sql.list_runs_for_job(job_id, limit).await
127    }
128
129    async fn list_runs_filtered(
130        &self,
131        job_id: Option<&str>,
132        status: Option<RunStatus>,
133        offset: usize,
134        limit: usize,
135    ) -> Result<Vec<Run>> {
136        self.sql
137            .list_runs_filtered(job_id, status, offset, limit)
138            .await
139    }
140
141    async fn claim_next_queued(
142        &self,
143        pool_id: &str,
144        worker_id: &str,
145        now: DateTime<Utc>,
146        lease_ttl_secs: i64,
147    ) -> Result<Option<Run>> {
148        claim::claim_next_queued(
149            &self.sql,
150            &self.redis,
151            pool_id,
152            worker_id,
153            now,
154            lease_ttl_secs,
155        )
156        .await
157    }
158
159    async fn claim_run_by_id(
160        &self,
161        run_id: &str,
162        pool_id: &str,
163        worker_id: &str,
164        now: DateTime<Utc>,
165        lease_ttl_secs: i64,
166    ) -> Result<Option<Run>> {
167        self.sql
168            .claim_run_by_id(run_id, pool_id, worker_id, now, lease_ttl_secs)
169            .await
170    }
171
172    async fn claim_runs_by_ids(
173        &self,
174        run_ids: &[&str],
175        pool_id: &str,
176        worker_id: &str,
177        now: DateTime<Utc>,
178        lease_ttl_secs: i64,
179    ) -> Result<Vec<Run>> {
180        self.sql
181            .claim_runs_by_ids(run_ids, pool_id, worker_id, now, lease_ttl_secs)
182            .await
183    }
184
185    async fn renew_run_lease(
186        &self,
187        run_id: &str,
188        worker_id: &str,
189        now: DateTime<Utc>,
190        lease_ttl_secs: i64,
191    ) -> Result<bool> {
192        self.sql
193            .renew_run_lease(run_id, worker_id, now, lease_ttl_secs)
194            .await
195    }
196
197    async fn reclaim_expired_run_leases(&self, now: DateTime<Utc>) -> Result<Vec<String>> {
198        let ids = self.sql.reclaim_expired_run_leases(now).await?;
199        for run_id in &ids {
200            if let Some(run) = self.sql.get_run(run_id).await? {
201                let pool = run_pool_key(run.pool_id.as_deref());
202                self.redis
203                    .enqueue_run(pool, &run.run_id, run.scheduled_for)
204                    .await?;
205            }
206        }
207        Ok(ids)
208    }
209
210    async fn append_revision(&self, revision: &chronon_core::models::JobRevision) -> Result<()> {
211        self.sql.append_revision(revision).await
212    }
213
214    async fn list_revisions(&self, job_id: &str) -> Result<Vec<chronon_core::models::JobRevision>> {
215        self.sql.list_revisions(job_id).await
216    }
217
218    async fn upsert_script(&self, script: &chronon_core::models::Script) -> Result<()> {
219        self.sql.upsert_script(script).await
220    }
221
222    async fn get_script(&self, script_name: &str) -> Result<Option<chronon_core::models::Script>> {
223        self.sql.get_script(script_name).await
224    }
225
226    async fn try_claim_run_once(
227        &self,
228        job_id: &str,
229        claimed_by: &str,
230        now: DateTime<Utc>,
231        claim_ttl_secs: i64,
232    ) -> Result<bool> {
233        self.sql
234            .try_claim_run_once(job_id, claimed_by, now, claim_ttl_secs)
235            .await
236    }
237
238    async fn mark_run_once_completed(
239        &self,
240        job_id: &str,
241        completed_at: DateTime<Utc>,
242    ) -> Result<()> {
243        self.sql.mark_run_once_completed(job_id, completed_at).await
244    }
245
246    async fn release_run_once_claim(
247        &self,
248        job_id: &str,
249        claimed_by: &str,
250        now: DateTime<Utc>,
251    ) -> Result<()> {
252        self.sql
253            .release_run_once_claim(job_id, claimed_by, now)
254            .await
255    }
256
257    async fn find_due_job_ids_in_partitions(
258        &self,
259        owned_partitions: &[u32],
260        due_until: DateTime<Utc>,
261        limit: u32,
262    ) -> Result<Vec<String>> {
263        self.sql
264            .find_due_job_ids_in_partitions(owned_partitions, due_until, limit)
265            .await
266    }
267
268    async fn min_next_run_at_in_partitions(
269        &self,
270        owned_partitions: &[u32],
271    ) -> Result<Option<DateTime<Utc>>> {
272        self.sql
273            .min_next_run_at_in_partitions(owned_partitions)
274            .await
275    }
276
277    async fn claim_job_for_tick(
278        &self,
279        job_id: &str,
280        claim_id: &str,
281        now: DateTime<Utc>,
282        lease_ttl_secs: i64,
283    ) -> Result<bool> {
284        self.sql
285            .claim_job_for_tick(job_id, claim_id, now, lease_ttl_secs)
286            .await
287    }
288
289    async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
290        self.sql.release_job_tick_claim(job_id).await
291    }
292
293    async fn persist_post_tick_job_state(
294        &self,
295        job_id: &str,
296        next_run_at: Option<DateTime<Utc>>,
297    ) -> Result<()> {
298        self.sql
299            .persist_post_tick_job_state(job_id, next_run_at)
300            .await
301    }
302
303    async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
304        self.sql.try_acquire_leader(instance_id, ttl_secs).await
305    }
306
307    async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
308        self.sql.renew_leader_lease(instance_id, ttl_secs).await
309    }
310
311    async fn get_leader(&self) -> Result<Option<chronon_core::models::SchedulerLeader>> {
312        self.sql.get_leader().await
313    }
314
315    async fn upsert_partition_assignment(
316        &self,
317        assignment: &chronon_core::models::PartitionAssignment,
318    ) -> Result<()> {
319        self.sql.upsert_partition_assignment(assignment).await
320    }
321
322    async fn list_partition_assignments(
323        &self,
324    ) -> Result<Vec<chronon_core::models::PartitionAssignment>> {
325        self.sql.list_partition_assignments().await
326    }
327
328    async fn register_worker(&self, worker: &chronon_core::models::Worker) -> Result<()> {
329        self.sql.register_worker(worker).await
330    }
331
332    async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
333        self.sql.heartbeat_worker(worker_id, at).await
334    }
335}
chronon_backend_redis/composite/mod.rs

chronon_backend_redis/composite/
mod.rs
