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 append_revision(&self, revision: &chronon_core::models::JobRevision) -> Result<()> {
198        self.sql.append_revision(revision).await
199    }
200
201    async fn list_revisions(&self, job_id: &str) -> Result<Vec<chronon_core::models::JobRevision>> {
202        self.sql.list_revisions(job_id).await
203    }
204
205    async fn upsert_script(&self, script: &chronon_core::models::Script) -> Result<()> {
206        self.sql.upsert_script(script).await
207    }
208
209    async fn get_script(&self, script_name: &str) -> Result<Option<chronon_core::models::Script>> {
210        self.sql.get_script(script_name).await
211    }
212
213    async fn try_claim_run_once(
214        &self,
215        job_id: &str,
216        claimed_by: &str,
217        now: DateTime<Utc>,
218        claim_ttl_secs: i64,
219    ) -> Result<bool> {
220        self.sql
221            .try_claim_run_once(job_id, claimed_by, now, claim_ttl_secs)
222            .await
223    }
224
225    async fn mark_run_once_completed(
226        &self,
227        job_id: &str,
228        completed_at: DateTime<Utc>,
229    ) -> Result<()> {
230        self.sql.mark_run_once_completed(job_id, completed_at).await
231    }
232
233    async fn release_run_once_claim(
234        &self,
235        job_id: &str,
236        claimed_by: &str,
237        now: DateTime<Utc>,
238    ) -> Result<()> {
239        self.sql
240            .release_run_once_claim(job_id, claimed_by, now)
241            .await
242    }
243
244    async fn find_due_job_ids_in_partitions(
245        &self,
246        owned_partitions: &[u32],
247        due_until: DateTime<Utc>,
248        limit: u32,
249    ) -> Result<Vec<String>> {
250        self.sql
251            .find_due_job_ids_in_partitions(owned_partitions, due_until, limit)
252            .await
253    }
254
255    async fn min_next_run_at_in_partitions(
256        &self,
257        owned_partitions: &[u32],
258    ) -> Result<Option<DateTime<Utc>>> {
259        self.sql
260            .min_next_run_at_in_partitions(owned_partitions)
261            .await
262    }
263
264    async fn claim_job_for_tick(
265        &self,
266        job_id: &str,
267        claim_id: &str,
268        now: DateTime<Utc>,
269        lease_ttl_secs: i64,
270    ) -> Result<bool> {
271        self.sql
272            .claim_job_for_tick(job_id, claim_id, now, lease_ttl_secs)
273            .await
274    }
275
276    async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
277        self.sql.release_job_tick_claim(job_id).await
278    }
279
280    async fn persist_post_tick_job_state(
281        &self,
282        job_id: &str,
283        next_run_at: Option<DateTime<Utc>>,
284    ) -> Result<()> {
285        self.sql
286            .persist_post_tick_job_state(job_id, next_run_at)
287            .await
288    }
289
290    async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
291        self.sql.try_acquire_leader(instance_id, ttl_secs).await
292    }
293
294    async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
295        self.sql.renew_leader_lease(instance_id, ttl_secs).await
296    }
297
298    async fn get_leader(&self) -> Result<Option<chronon_core::models::SchedulerLeader>> {
299        self.sql.get_leader().await
300    }
301
302    async fn upsert_partition_assignment(
303        &self,
304        assignment: &chronon_core::models::PartitionAssignment,
305    ) -> Result<()> {
306        self.sql.upsert_partition_assignment(assignment).await
307    }
308
309    async fn list_partition_assignments(
310        &self,
311    ) -> Result<Vec<chronon_core::models::PartitionAssignment>> {
312        self.sql.list_partition_assignments().await
313    }
314
315    async fn register_worker(&self, worker: &chronon_core::models::Worker) -> Result<()> {
316        self.sql.register_worker(worker).await
317    }
318
319    async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
320        self.sql.heartbeat_worker(worker_id, at).await
321    }
322}
chronon_backend_redis/composite/mod.rs

chronon_backend_redis/composite/
mod.rs
