Stephan Deumier
Repositories
7
 Projects Packages
25
 Public activity Starred repositories

pg-jobs (0.8.0)

Published 2026-07-05 20:05:50 +02:00 by AzSiAz in AzSiAz/pg-jobs

Installation

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

[net]
git-fetch-with-cli = true
cargo add pg-jobs@0.8.0 --registry forgejo

About this package

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, per-node dedup keys, and per-node priority (with a flow-wide default_priority fallback). Resume a failed flow in place with requeue_flow — it re-drives the dead/cancelled subgraph and recomputes dependents, instead of cloning the whole flow like rerun_flow.
  • Dynamic joins — declare a rendezvous job up front (root baton: Deps::on_self() from the coordinator), then let any job holding a still-unsatisfied edge into it grow its prerequisite set at runtime with spawn_into / fan_out_into (+ _to variants). The held edge keeps the join unclaimable mid-growth, so a runtime-discovered fan (per-device chunks, per-item chains) can feed a single rollup that starts only when the whole dynamic subtree is terminal — and dependencies_of still delivers every leaf's typed output to it.
  • 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.
  • Per-kind panic policy — Job::panic_policy() (Retryable by default, or Fatal). A handler panic is caught and recorded immediately, releasing the lease without waiting for the reaper: Retryable burns an attempt, Fatal dead-letters at once.
  • Idle execution timeout — a renewable per-queue watchdog (derived from the queue's reaper lock window) fails a handler that goes silent (no ctx.heartbeat/progress) too long, cancelling it and freeing the worker slot. A long-running job must call ctx.heartbeat() periodically; the detached background heartbeat keeps the DB lease fresh for the reaper but does not reset this watchdog (a non-yielding blocking handler still can't be interrupted — a Tokio limitation).
  • 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.
  • Synchronous / blocking jobs — implement JobSync (a plain fn, run on spawn_blocking) for CPU-bound or blocking work; register_sync + enqueue_sync. Cooperative cancellation via ctx.heartbeat() / ctx.is_cancelled() (a silent handler past the idle timeout is dead-lettered without requeue).
  • Basic jobs (high-throughput tier) — implement BasicJob for many short, idempotent tasks. The worker serves dedicated basic_queues with a batched claim + pipelined execution
    • an interval-flushed bulk completer, trading the rich tier's transaction/Output/flows/caps for throughput. Delivery is at-least-once and there is no heartbeat — the lease is the queue's lock_duration, so keep handlers under it. register_basic + enqueue_basic:
    registry.register_basic(MyJob);
let config = WorkerConfig::default().with_basic_queues([Q::Fast]);
store.enqueue_basic_to::<MyJob>(input, Q::Fast).await?;
  • Debounce enqueue — enqueue_debounced (+ _basic/_sync and _to variants) coalesce a burst sharing a dedup_key into one pending job carrying the latest payload/run-time (replace-if-pending), reusing the dedup index with no migration.
  • Recurring scheduler — fixed-interval (upsert_every::<J>) and cron (upsert_cron::<J>, standard 5-field UTC) schedules, typed by job (kind and the payload template come from J + a typed input), keyed by name for idempotent startup declaration; each fired job is labelled by J::name(&input) (which defaults to the kind) for dashboards; 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_*).
  • Pause / resume a queue — pause_queue/resume_queue flip a queue_config flag that claim_next honours fleet-wide (in-flight jobs keep running; new jobs still enqueue and run once resumed). A NOTIFY wakes idle workers on the change, and is_queue_paused reads the state for dashboards.
  • Per-key concurrency — a job's Job::group_key(&input) (derived from the typed input) caps how many jobs of the same (queue, group_key) run at once, so a fixed slot pool spreads across distinct keys (e.g. scrape ≤1 per source) instead of letting one key monopolise it. Cap via set_queue_group_concurrency (per-queue default) or set_group_concurrency (per-key override); orthogonal to max_concurrency. The per-key gates are hard fleet-wide guarantees (claimers serialize per key on a transaction-scoped advisory lock at claim time; busy claimers divert to another group rather than wait) — suitable for crawl politeness; queue-level caps remain advisory.
  • Maintenance jobs — garbage collection (GcJob) and the metrics rollup (MetricsRollupJob) ship as built-in detached jobs you register and schedule with the Scheduler, each advisory-locked so a single worker runs it under many. Retention-based GC deletes terminated jobs past their (per-queue overridable) windows; the rollup precomputes metrics decoupled from that retention. The JobStore::gc / rollup_sweep primitives remain callable directly.
  • Tracking — read API for flow views, queue stats, job listing, and queue_metrics (time-bucketed terminal-status counts for dashboards, served from the rollup for settled history and stitched with a live tail, zero-filled).
  • Display names — free-form, cosmetic labels per job (JobSpec::name / Job::name(&input)) and per flow (Flow::new(name) / Flow::node_name), surfaced by the tracking API (FlowSummary.name, JobRow.name) without touching routing or dedup.
  • Job admin — cancel or remove a job by id (cascading to its dependents), read a job's log lines, re-drive a single dead job or a whole queue's dead-letter at once (cancel, remove, job_logs, requeue, requeue_dead).
  • Bulk enqueue — enqueue_specs/enqueue_specs_on insert many jobs in one multi-row INSERT (ids in input order); enqueue_many::<J>/enqueue_many_to/ enqueue_many_on are the typed conveniences (per-input group_key/defaults derived via for_job; _to for an explicit queue, _on within a transaction).
  • 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.8", 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

An async Job runs in one of two modes, chosen by Job::mode() (default Transactional); a synchronous, blocking handler uses the JobSync trait instead (see Synchronous / blocking jobs):

  • 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.

    Note: that background heartbeat keeps the DB lease fresh for the reaper, but it does not reset the idle execution-timeout watchdog (see Features). A handler that stays silent (no ctx.heartbeat()/progress) longer than the queue's idle timeout is cancelled and fails as retryable — call ctx.heartbeat() periodically during long silent work.

impl Job for ScrapeChapter {
    fn mode() -> ExecMode { ExecMode::Detached }
    async fn run(&self, input: In, ctx: &mut Ctx<'_>) -> Result<Out, JobError> {
        for page in pages {
            fetch_and_upload(page).await?;                       // long, silent I/O
            ctx.heartbeat().await.map_err(JobError::retryable)?; // re-arm the watchdog
        }
        Ok(out)
    }
}

Synchronous / blocking jobs (JobSync)

For CPU-bound or blocking work that can't be made async, implement JobSync instead of Job: its run is a plain fn, executed on a spawn_blocking thread so it doesn't starve the runtime. Register with Registry::register_sync and enqueue with store.enqueue_sync::<J>(input). It is at-least-once and gets a SyncCtx with no database access — do your own synchronous writes, or none.

A blocking thread can't be forcibly cancelled, so the contract is cooperative:

  • call ctx.heartbeat() periodically — a handler that stays silent past the queue's idle timeout is dead-lettered without requeue (its thread is detached, so it would otherwise leak), and
  • poll ctx.is_cancelled() between work units and return early when set (on idle timeout, or when the job is cancelled/reaped elsewhere) to free the thread promptly.
impl JobSync for ParsePdf {
    type Input = Doc;
    type Output = Pages;
    fn run(&self, doc: Doc, ctx: &mut SyncCtx) -> Result<Pages, JobError> {
        let mut pages = Vec::new();
        for page in doc.pages() {
            if ctx.is_cancelled() {
                return Err(JobError::retryable("cancelled"));
            }
            pages.push(parse_blocking(page)?); // blocking CPU work
            ctx.heartbeat();                   // prove liveness (never blocks)
        }
        Ok(pages)
    }
}

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.

Benchmark & demo app

The benchmark/ workspace crate is both a performance harness and a reference demo: twelve scenarios exercise the engine end to end (DAG flows, fan-out, retry/DLQ, deferrals, concurrency caps, rate limits, pause/resume, the scheduler, GC/rollup), measure sustained throughput and latency percentiles, and assert observed behaviour with pass/fail checks. Each run emits a self-contained HTML report plus a JSON result, and regenerates the tracked feature-gap document at docs/ecarts.html.

docker compose up -d
export DATABASE_URL="postgres://postgres:postgres@localhost:5432/pg_jobs"
cargo run -p benchmark            # ⚠️ wipes the `jobs` schema at DATABASE_URL

See benchmark/README.md for profiles, flags, artefacts, and how to read the report.

License

MIT © Stephan Deumier

Dependencies

ID Version
anyhow ^1
async-trait ^0.1
chrono ^0.4
croner ^2
futures-util ^0.3
pg-jobs-derive =0.8.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-07-05 20:05:50 +02:00
10
MIT
472 KiB
Assets (1)
pg-jobs-0.8.0.crate 472 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