Struct PostgresRedisSchedulerStore 

pub struct PostgresRedisSchedulerStore { /* private fields */ }

SQL persistence with Redis-backed run claim ordering.

Production Mode 2 store: job/run rows stay in Postgres (or any SchedulerStore); workers claim through Redis ZSETs for higher throughput. Requires facade features postgres and redis. Build with Self::new after connecting both layers.

Mode 2 examples: coordinator / worker.

See crate::RedisQueueLayer for URL / prefix / cluster env vars (CHRONON_REDIS_URL, CHRONON_REDIS_CLUSTER_URLS, CHRONON_REDIS_HASH_TAGS).

Runnable: cargo run -p uf-chronon --example postgres_redis_boot --features postgres,redis and the coordinator_daemon / worker_daemon pair.

Implementations

impl PostgresRedisSchedulerStore

pub fn new( sql: Arc<dyn SchedulerStore>, redis: RedisQueueLayer, ) -> PostgresRedisSchedulerStore

Wrap a SQL store and Redis queue layer.

create_run writes SQL then enqueues Redis for queued runs; claim_next_queued pops Redis then updates SQL lease state.

Examples

use std::sync::Arc;

use chronon_backend_postgres::PostgresSchedulerStore;
use chronon_backend_redis::{PostgresRedisSchedulerStore, RedisQueueLayer};
use chronon_core::store::SchedulerStore;

let sql: Arc<dyn SchedulerStore> = Arc::new(
    PostgresSchedulerStore::connect("postgres://localhost/chronon").await?,
);
let redis = RedisQueueLayer::connect("redis://127.0.0.1:6379", Some("myapp")).await?;
let store = PostgresRedisSchedulerStore::new(sql, redis);

Trait Implementations

impl Debug for PostgresRedisSchedulerStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl SchedulerStore for PostgresRedisSchedulerStore

fn upsert_job<'life0, 'life1, 'async_trait>( &'life0 self, job: &'life1 Job, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Insert or replace a job row keyed by Job::job_id.

fn get_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Look up a job by primary key.

fn get_job_by_name<'life0, 'life1, 'async_trait>( &'life0 self, job_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Look up a job by unique Job::job_name.

fn list_jobs<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Return all jobs (admin / list API).

fn list_due_jobs<'life0, 'async_trait>( &'life0 self, before: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Jobs with next_run_at <= before and enabled scheduling (tick discovery).

fn pause_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Disable automatic scheduling without deleting the job.

fn resume_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Re-enable automatic scheduling after Self::pause_job.

fn create_run<'life0, 'life1, 'async_trait>( &'life0 self, run: &'life1 Run, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Persist a new run row (typically RunStatus::Queued).

fn update_run<'life0, 'life1, 'async_trait>( &'life0 self, run: &'life1 Run, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Replace an existing run row (status transitions, lease renewal, completion).

fn get_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Look up a run by Run::run_id.

fn list_runs_for_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Recent runs for one job, newest first, capped by limit.

fn list_runs_filtered<'life0, 'life1, 'async_trait>( &'life0 self, job_id: Option<&'life1 str>, status: Option<RunStatus>, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Paginated run listing with optional job and status filters (HTTP list API).

fn claim_next_queued<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, pool_id: &'life1 str, worker_id: &'life2 str, now: DateTime<Utc>, lease_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<Option<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Claim the next queued run for a worker pool.

fn claim_run_by_id<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, run_id: &'life1 str, pool_id: &'life2 str, worker_id: &'life3 str, now: DateTime<Utc>, lease_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<Option<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Claim a specific queued run by id (postgres-redis hybrid hot path).

fn claim_runs_by_ids<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, run_ids: &'life1 [&'life2 str], pool_id: &'life3 str, worker_id: &'life4 str, now: DateTime<Utc>, lease_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Claim multiple queued runs by id in one round trip when the backend supports it.

fn renew_run_lease<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, worker_id: &'life2 str, now: DateTime<Utc>, lease_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<bool, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Extend the worker lease on a claimed run if worker_id still holds the claim.

fn append_revision<'life0, 'life1, 'async_trait>( &'life0 self, revision: &'life1 JobRevision, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Append an immutable job revision snapshot (audit / rollback).

fn list_revisions<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<JobRevision>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

All revisions for a job, typically oldest-first.

fn upsert_script<'life0, 'life1, 'async_trait>( &'life0 self, script: &'life1 Script, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Insert or replace script metadata (name, signature hash).

fn get_script<'life0, 'life1, 'async_trait>( &'life0 self, script_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Script>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Look up persisted script metadata by name.

fn try_claim_run_once<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: &'life1 str, claimed_by: &'life2 str, now: DateTime<Utc>, claim_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<bool, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Attempt exclusive claim for a run-once job before enqueueing its single run.

fn mark_run_once_completed<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, completed_at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Mark a run-once job as finished so future ticks skip it.

fn release_run_once_claim<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: &'life1 str, claimed_by: &'life2 str, now: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Release a run-once claim when enqueue failed or the coordinator shut down cleanly.

fn find_due_job_ids_in_partitions<'life0, 'life1, 'async_trait>( &'life0 self, owned_partitions: &'life1 [u32], due_until: DateTime<Utc>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Due job ids owned by this coordinator’s partition slice (distributed tick).

fn min_next_run_at_in_partitions<'life0, 'life1, 'async_trait>( &'life0 self, owned_partitions: &'life1 [u32], ) -> Pin<Box<dyn Future<Output = Result<Option<DateTime<Utc>>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Earliest next_run_at among jobs in the owned partitions (sleep hint for tick loop).

fn claim_job_for_tick<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: &'life1 str, claim_id: &'life2 str, now: DateTime<Utc>, lease_ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<bool, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Exclusive short-lived lease on a job row during tick processing.

fn release_job_tick_claim<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Release tick claim after enqueue succeeds or the tick aborts.

fn persist_post_tick_job_state<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, next_run_at: Option<DateTime<Utc>>, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Persist next_run_at (and related job fields) after a successful tick.

fn try_acquire_leader<'life0, 'life1, 'async_trait>( &'life0 self, instance_id: &'life1 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<bool, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Attempt to become the active scheduler leader (singleton row + TTL).

fn renew_leader_lease<'life0, 'life1, 'async_trait>( &'life0 self, instance_id: &'life1 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Renew the leader lease while this instance remains leader.

fn get_leader<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<SchedulerLeader>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Current leader row, if any (expired leases may still be returned for diagnostics).

fn upsert_partition_assignment<'life0, 'life1, 'async_trait>( &'life0 self, assignment: &'life1 PartitionAssignment, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Upsert partition ownership for coordinator sharding.

fn list_partition_assignments<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<PartitionAssignment>, ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

All partition assignments (rebalance / diagnostics).

fn register_worker<'life0, 'life1, 'async_trait>( &'life0 self, worker: &'life1 Worker, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Register or update a worker heartbeat row.

fn heartbeat_worker<'life0, 'life1, 'async_trait>( &'life0 self, worker_id: &'life1 str, at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), ChrononError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresRedisSchedulerStore: 'async_trait,

Update last_heartbeat_at for an existing worker.