Struct PostgresSchedulerStore 

pub struct PostgresSchedulerStore { /* private fields */ }

PostgreSQL-backed scheduler store.

Shared durable storage for Mode 2 coordinator–worker clusters (and Mode 1 when you already run Postgres). Pass a connection URL to Self::connect; daemons often use CHRONON_POSTGRES_URL / postgres_test_url.

Mode 2 examples: coordinator / worker.

For higher claim throughput, wrap with PostgresRedisSchedulerStore from chronon-backend-redis (postgres + redis features). Enable the facade postgres feature to re-export this type.

Examples

use chronon_backend_postgres::PostgresSchedulerStore;

let store = PostgresSchedulerStore::connect(
    "postgres://user:pass@localhost/chronon",
)
.await?;

Runnable: cargo run -p uf-chronon --example postgres_boot --features postgres.

Implementations

impl PostgresSchedulerStore

pub async fn connect(url: &str) -> Result<PostgresSchedulerStore, ChrononError>

Connect using a PostgreSQL connection URL.

Examples

use chronon_backend_postgres::PostgresSchedulerStore;

let store = PostgresSchedulerStore::connect(
    "postgres://user:pass@localhost/chronon",
)
.await?;

Errors

Returns a storage error when the pool cannot connect or schema bootstrap fails.

pub async fn connect_isolated( url: &str, schema: &str, ) -> Result<PostgresSchedulerStore, ChrononError>

Connect with an isolated schema (for parallel tests).

Examples

use chronon_backend_postgres::PostgresSchedulerStore;

let store = PostgresSchedulerStore::connect_isolated(
    "postgres://user:pass@localhost/chronon",
    "chronon_test_schema",
)
.await?;

Errors

Returns a storage error when schema creation, pool connect, or bootstrap fails.

pub async fn attach_isolated( url: &str, schema: &str, ) -> Result<PostgresSchedulerStore, ChrononError>

Attach to an existing isolated schema (no DDL bootstrap; for multi-process workers).

Errors

Returns a storage error when the pool cannot be opened.

pub async fn drop_isolated_schema( url: &str, schema: &str, ) -> Result<(), ChrononError>

Drop an isolated bench/test schema (multibench cell reset).

Errors

Returns a storage error when the admin connection or DDL fails.

pub async fn from_pool( pool: Pool<Postgres>, ) -> Result<PostgresSchedulerStore, ChrononError>

Wrap an existing pool (schema bootstrap runs).

Examples

use chronon_backend_postgres::PostgresSchedulerStore;
use sqlx::postgres::PgPoolOptions;

let pool = PgPoolOptions::new()
    .connect("postgres://localhost/chronon")
    .await
    .expect("pool");
let store = PostgresSchedulerStore::from_pool(pool).await?;

Errors

Returns a storage error when schema bootstrap fails.

pub fn pool(&self) -> &Pool<Postgres>

Underlying connection pool.

Panics

Panics if the inner pool is not PostgreSQL (internal invariant violation).

Trait Implementations

impl Debug for PostgresSchedulerStore

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

Formats the value using the given formatter.

impl SchedulerStore for PostgresSchedulerStore

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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: '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, PostgresSchedulerStore: 'async_trait,

Update last_heartbeat_at for an existing worker.