Crate cronoq_store_postgres 

cronoq-store-postgres

A PostgreSQL Store for cronoq with active-passive high availability.

• Write-ahead state: every engine state transition is committed in batches, one transaction per batch, before submitters are acknowledged.
• Isolation: each scheduler cluster lives in its own PostgreSQL schema (PgStoreConfig::schema); clusters in different schemas share nothing, not even locks.
• Leader election: a dedicated session holds an advisory lock derived from the schema; acquiring it bumps a fencing epoch in cronoq_leader.
• Fencing: each commit re-reads the epoch in the same transaction and fails with StoreError::Fenced if another instance took over.
• Recovery: a new leader loads live tasks, recently finished ones (for key deduplication), every finished task a live task still depends on, and recurring schedules. Rows that cannot be decoded are quarantined (marked failed) instead of blocking recovery.
• Errors: connection loss, timeouts, serialization failures and server shutdowns are reported as StoreError::Unavailable, which the scheduler retries.

Times are stored as wall-clock Unix nanoseconds. PostgreSQL cannot store NUL characters in text, so NUL in strings (messages, keys, tags, …) is stored as U+FFFD.

use bytes::Bytes;
use cronoq::{Pool, Scheduler};
use cronoq_store_postgres::PgStore;

let store = PgStore::connect("postgres://localhost/cronoq").await?;
let sched = Scheduler::<Bytes, Bytes>::builder()
    .pool(Pool::new("default").executor_fn(|_ctx, p: std::sync::Arc<Bytes>| async move {
        Ok::<_, cronoq::TaskError>((*p).clone())
    }))
    .store(store.clone())
    .build() // blocks until this instance is the leader
    .await?;
tokio::select! {
    _ = store.leadership_lost() => { /* step down: stop serving, exit or rejoin */ }
    _ = tokio::signal::ctrl_c() => {}
}

Structs

PgStore

PostgreSQL-backed Store. Cheap to clone.

PgStoreConfig

Store configuration.

Type Aliases

TaskRecord

Persisted view of a task, for lookups by id or key.