pg-jobs (0.1.0)

Published 2026-06-28 17:37:01 +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.1.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; fatal vs. retryable errors.
Transactional execution — handlers can run inside the job's own transaction (ExecMode::Transactional), committing work atomically with completion.
Recurring scheduler — every_ms schedules driven from the worker loop.
Tracking — read API for flow views, queue stats, and job listing.
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.1", 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, JobError, JobSpec, Registry, Worker, WorkerConfig};
use pg_jobs::ctx::Ctx;
use pg_jobs::store::JobStore;
use std::sync::Arc;
use std::time::Duration;

// 1. Define a typed job.
struct SendEmail;

#[async_trait::async_trait]
impl Job for SendEmail {
    const KIND: &'static str = "send_email";
    const QUEUE: &'static str = "default";
    type Input = String;   // recipient address
    type Output = ();

    async fn run(&self, to: String, _ctx: &mut Ctx<'_>) -> Result<(), JobError> {
        // ... do the work; map errors with JobError::retryable / JobError::fatal
        println!("sending to {to}");
        Ok(())
    }
}

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

    // 2. Apply migrations.
    pg_jobs::MIGRATOR.run(&pool).await?;

    // 3. Register handlers and start a worker.
    let store = JobStore::new(pool.clone());
    let mut reg = Registry::new();
    reg.register(SendEmail);

    let worker = Worker::new(
        store.clone(),
        Arc::new(reg),
        WorkerConfig {
            queues: vec!["default".into()],
            poll_interval: Duration::from_millis(500),
            ..Default::default()
        },
    );

    // 4. Enqueue work.
    store
        .enqueue(JobSpec::for_job::<SendEmail>("alice@example.com".into())?)
        .await?;

    // 5. Run until a shutdown signal is received.
    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.

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
serde	^1
serde_json	^1
sqlx	^0.9
thiserror	^2
tokio	^1
tracing	^0.1
uuid	^1
sqlx	^0.9
tokio	^1

Details

Cargo

AzSiAz/pg-jobs

2026-06-28 17:37:01 +02:00

MIT

80 KiB

Assets (1)

pg-jobs-0.1.0.crate 80 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