pg-jobs (0.2.0)

Published 2026-06-29 02:00:44 +02:00 by AzSiAz in AzSiAz/pg-jobs

Setup this registry in the Cargo configuration file (for example ~/.cargo/config.toml):

[registries.forgejo]
index = "sparse+" # Sparse index
# index = "" # Git

[net]
git-fetch-with-cli = true

To install the package using Cargo, run the following command:

cargo add pg-jobs@0.2.0 --registry forgejo

For more information on the Cargo registry, see the documentation.

Postgres-native job/flow engine (DAG flows, retry/DLQ, typed handlers)

pg-jobs

A Postgres-native job & flow engine for Rust: DAG flows, retry/DLQ, typed handlers, recurring scheduler, and a worker driven by LISTEN/NOTIFY (with a polling fallback). All state lives in a dedicated jobs schema in your own Postgres database — no external broker.

Features

Typed handlers — implement the Job trait with typed Input/Output.
DAG flows — compose jobs with dependencies (Flow, submit_flow, rerun_flow), fan-out, and per-node dedup keys.
Retry / DLQ — configurable backoff strategies (exponential/fixed, optional equal jitter); fatal vs. retryable errors, plus JobError::Delayed to postpone a job (e.g. on a 429) without burning an attempt, pairing with ctx.update_data.
Two execution modes — Transactional (default) commits a handler's work atomically with completion; Detached holds no transaction (at-least-once) for long, I/O-bound jobs. See Execution modes.
Recurring scheduler — fixed-interval (upsert_every) and cron (upsert_cron, standard 5-field UTC) schedules, keyed by name for idempotent startup declaration; remove_schedule to drop one. Driven from the worker loop.
Per-queue tuning — concurrency cap, throughput rate limit (max starts per sliding window), reaper lock duration, and GC retention, each overridable per queue (set_queue_*).
Automatic GC — opt-in retention-based cleanup driven by the worker (Worker::with_gc), advisory-locked so a single worker runs it under many.
Tracking — read API for flow views, queue stats, and job listing.
Job admin — cancel or remove a job by id (cascading to its dependents), and read a job's log lines (cancel, remove, job_logs).
No compile-time DB — uses runtime sqlx queries, so it builds without a DATABASE_URL or .sqlx cache.

Install

Released versions are published to the Forgejo Cargo registry. Declare the registry in your project's .cargo/config.toml:

[registries.forgejo]
index = "sparse+https://forge.netserv.fr/api/packages/AzSiAz/cargo/"

Then depend on it in Cargo.toml:

[dependencies]
pg-jobs = { version = "0.2", registry = "forgejo" }

Authenticate once for read access with cargo login --registry forgejo (paste a Forgejo token in the Bearer <token> form when prompted).

Alternatively, track main directly via git:

[dependencies]
pg-jobs = { git = "ssh://git@git.netserv.fr/AzSiAz/pg-jobs.git", branch = "main" }

The library crate is imported as pg_jobs.

Migrations

The crate ships its schema migrations and exposes a Migrator. Run it once at startup; the jobs schema is created automatically.

pg_jobs::MIGRATOR.run(&pool).await?;

Usage

use pg_jobs::{Job, JobKind, JobError, JobSpec, QueueName, Registry, Worker, WorkerConfig};
use pg_jobs::ctx::Ctx;
use pg_jobs::store::JobStore;
use std::sync::Arc;

// 0. Declare your queues once — typo-safe, autocompleted.
#[derive(Clone, Copy, QueueName)]
enum Queue {
    Default,
    Urgent,
}

// 1. Define a typed job. `KIND` is derived from the struct name ("send_email").
#[derive(JobKind)]
struct SendEmail;

#[async_trait::async_trait]
impl Job for SendEmail {
    type Input = String;   // recipient address
    type Output = ();

    async fn run(&self, to: String, _ctx: &mut Ctx<'_>) -> Result<(), JobError> {
        println!("sending to {to}");
        Ok(())
    }

    // Per-kind retry policy. These are associated functions (no `self`), like
    // `mode()`. `JobSpec::for_job` seeds every enqueue with these defaults.
    fn max_attempts() -> u32 {
        3
    }
    fn backoff() -> pg_jobs::Backoff {
        pg_jobs::Backoff::exponential(
            std::time::Duration::from_secs(1),
            std::time::Duration::from_secs(30),
        )
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = sqlx::PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    pg_jobs::MIGRATOR.run(&pool).await?;

    // Jobs without an explicit queue land on `Queue::Default`.
    let store = JobStore::with_default_queue(pool.clone(), Queue::Default);
    let mut reg = Registry::new();
    reg.register(SendEmail);

    let worker = Worker::new(
        store.clone(),
        Arc::new(reg),
        WorkerConfig::default().with_queues([Queue::Default, Queue::Urgent]),
    );

    // Enqueue on the default queue…
    store.enqueue::<SendEmail>("alice@example.com".into()).await?;
    // …or route per call (e.g. premium users).
    store.enqueue_to::<SendEmail>("vip@example.com".into(), Queue::Urgent).await?;
    // …or build a spec for finer control (priority, delay, dedup, retry overrides).
    store
        .enqueue_spec(
            JobSpec::for_job::<SendEmail>("bob@example.com".into())?
                .on_queue(Queue::Urgent)
                .priority(10)
                .delay(std::time::Duration::from_secs(30))
                .dedup_key("welcome:bob@example.com"),
        )
        .await?;

    let (_tx, rx) = tokio::sync::watch::channel(false);
    worker.run(rx).await?;
    Ok(())
}

See the crate's tests/ directory for flows, fan-out, transactional jobs, recurring schedules, the reaper, and tracking.

Execution modes

Each job kind runs in one of two modes, chosen by Job::mode() (default Transactional):

Transactional — the handler runs inside the job's transaction. Writes on it via ctx.tx() commit atomically with the job's completion, so a crash leaves neither — this is what removes the dual-write races a Redis-backed queue is prone to. The trade-off: the job pins a pooled connection for its whole duration, so keep these jobs short. Best when the work is a database write.
Detached — the handler runs with no open job transaction, while a background heartbeat keeps the lock fresh. Completion is a separate guarded step, so delivery is at-least-once: make these handlers idempotent. Best for long, I/O-bound work (HTTP scraping, S3 uploads) where holding a transaction open would cause idle-in-transaction bloat and pin a connection.

impl Job for ScrapeChapter {
    // ... long HTTP fetch + S3 upload ...
    fn mode() -> ExecMode { ExecMode::Detached }
}

Writing to the database from a handler

Two helpers, and only tx() is mode-specific:

ctx.with_tx(|conn| …) — works in both modes: a short transaction that commits immediately, independent of the job. Use it for incremental, idempotent writes (and it's the portable choice if a handler might switch modes).
ctx.tx() — Transactional only: hands you the job's own transaction, so the writes commit atomically with completion. In Detached mode there is no job transaction, so it returns an error — use with_tx() there.

So reach for tx() only when you specifically want atomic-with-completion; otherwise with_tx() is the uniform API across both modes.

Concurrency and connections

WorkerConfig::max_in_flight sets how many jobs one worker runs at once. In Transactional mode each in-flight job pins a pooled connection for its whole run, so size the sqlx pool to at least max_in_flight (+1 for claims) or jobs will block on pool.begin(). Detached jobs only touch a connection briefly (claim, heartbeat, completion), so a small pool serves high concurrency.

A connection pooler like PgBouncer caps total Postgres backends across many worker processes, but won't shrink what a single worker needs for concurrent Transactional jobs: a held transaction can't be multiplexed. Note that PgBouncer's transaction pooling disables LISTEN/NOTIFY (the worker falls back to polling) and needs prepared-statement-aware settings; pg-jobs' advisory locks are transaction-scoped and stay compatible.

Development

A compose.yaml provides a local Postgres 18 (with extensions baked in) for running the test suite:

docker compose up -d
export DATABASE_URL="postgres://postgres:postgres@localhost:5432/pg_jobs"
cargo test

The integration tests use #[sqlx::test], which connects to the base database and creates/drops an isolated temporary database per test — so the configured role must be a superuser (the default postgres user is). The crate builds without a database (cargo build / cargo check); only the tests need one.

License

MIT © Stephan Deumier

ID	Version
anyhow	^1
async-trait	^0.1
chrono	^0.4
croner	^2
pg-jobs-derive	=0.2.0
rand	^0.10
serde	^1
serde_json	^1
sqlx	^0.9
thiserror	^2
tokio	^1
tracing	^0.1
uuid	^1
sqlx	^0.9
tokio	^1
trybuild	^1

Details

Cargo

AzSiAz/pg-jobs

2026-06-29 02:00:44 +02:00

MIT

118 KiB

Assets (1)

pg-jobs-0.2.0.crate 118 KiB

Versions (16) View all

0.10.0

2026-07-21

0.9.6

2026-07-11

0.9.5

2026-07-11

0.9.4

2026-07-09

0.9.3

2026-07-07

Issues