pg-jobs (0.10.0)

Published 2026-07-21 00:48:39 +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.10.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, 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 runtime 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).
Supervised async execution — every handler runs without a handler-wide transaction; the runtime owns heartbeat and short fenced settlement. Ctx exposes bounded queue capabilities, never the worker pool. Delivery is at-least-once, so handlers must be idempotent. See Execution contract.
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 through the common supervised executor and a batch-oriented settlement lane, trading the rich tier's Output/flows/caps for throughput. Execution slots and pending settlements are bounded separately so batches can fill without starving handlers. Delivery can repeat within the configured business-attempt budget and the bounded infrastructure-refund budget, and there is no heartbeat — the lease is the queue's lock_duration, so keep handlers under it. With the default max_attempts = 1, a stale first lease is dead-lettered instead of replayed; a transaction-abort refund remains a separate bounded recovery path. Raise the business budget when crash replay is required. A non-flow Basic job may still be a prerequisite; its successful batch flush resolves outgoing edges atomically. 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 by the daemon's supervised scheduler service.
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 idempotent 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.10.0", 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?;

Optional restricted runtime role

Nothing else is required for the normal single-owner setup above. Applications that intentionally connect at runtime through an already provisioned, restricted PostgreSQL login may opt in to pg-jobs ACL reconciliation after the owner has run migrations:

use pg_jobs::access::{AccessProfile, RoleGrant};

pg_jobs::MIGRATOR.run(&owner_pool).await?;
pg_jobs::access::reconcile(
    &owner_pool,
    &[RoleGrant::runtime("my_runtime_role")],
)
.await?;

pg_jobs::access::validate_current(&runtime_pool, AccessProfile::Runtime).await?;

Repeat that order after every pg-jobs schema migration: migrate with the owner pool, rerun access::reconcile, then run access::validate_current through the restricted runtime pool before starting workers. The Runtime profile receives EXECUTE only on the exact thirteen supported functions in jobs_api_v1 and on no function in jobs; the legacy jobs.resolve_dependency_cascade(uuid[]) primitive and the internal trigger functions are not exposed for direct calls. Mutating lifecycle calls require the connection to actually use READ COMMITTED (PostgreSQL's default) and a whole-transaction retry after 40P01 or 40001. ACL reconciliation and validate_current validate privileges, not the application's transaction isolation; callers must enforce this precondition on every relevant connection.

The separate v1 schema specifically keeps the published 0.9.6 ACL reconciler, which requires zero executable functions in jobs and ignores jobs_api_v1, from revoking the new API. The final Runtime profile deliberately satisfies that same zero-function jobs contract. It is not an ABI bridge for intermediate development builds: builds whose exact allow-list contains only the first eight, nine or ten v1 signatures are incompatible with migration 0033's final thirteen-signature surface (including fenced group deferral and the three re-drive entry points) and may reject or revoke newer functions. Deploy migrations, reconciler and workers from the same final release.

Migration 0032 also scopes transaction-abort release markers to an explicit lifecycle generation. Migration 0033 introduces jobs_api_v1.redrive_job, the set-wise jobs_api_v1.redrive_dead, and complete jobs_api_v1.redrive_flow. These functions own the terminal-state filter, budget reset, generation advance and transactional queue wakeup. The flow primitive additionally owns its busy fence, dependency recount and relative-delay restoration. Migration 0033 serializes membership INSERTs against that re-drive with shared/exclusive transaction advisory locks and makes flow_id immutable after insertion, so direct SQL and a future Go worker receive the same closed-set boundary without a PostgreSQL extension. The final Rust requeue, requeue_dead, and requeue_flow methods call those same portable boundaries that a future Go worker can use. During a rolling handover, deploy migration, final ACL reconciler and callers together; an intermediate binary or a hand-written SQL status='pending', attempts=0 update still bypasses that contract. Newly inserted jobs and full rerun_flow clones start at generation zero normally.

Run redrive_flow in autocommit, or commit/rollback its transaction immediately after the call even when it returns busy=true: its advisory fence is transaction-scoped and otherwise blocks new flow members until the surrounding transaction ends. Retry the whole administrative call after 40P01 or 40001. Before production rollout, rehearse migration 0032 on a production-sized copy; its job_attempt backfill runs while migration locks are held, so use a bounded lock_timeout and a controlled retry or maintenance window rather than waiting unboundedly on live traffic.

That ACL compatibility also does not make graph mutations safe across arbitrary old/new binaries. Stop or replace legacy dependency producers, Basic completers, and administrative remove/purge callers before enabling v1 graph writers and terminal transitions on the same graphs. Do not let an old Basic completer finish a Basic prerequisite during that handover. Existing terminal Basic prerequisites may already have blocked dependents; audit them separately instead of blindly replaying resolve_dependents, which is not idempotent on an already-resolved edge. In the upgraded Rust worker, a flow-bearing Basic job uses the unitary commit_completion pipeline backed by the measured UPDATE/probe/resolver split; only non-flow Basic successes use complete_basic_jobs. A portable worker should start with the holistic complete_job function for that unitary case and copy the Rust split only after its own benchmark justifies it.

RoleGrant never creates or alters PostgreSQL roles, credentials or memberships. It only reconciles ACLs on pg-jobs business objects. The runtime profile has USAGE (without CREATE) on jobs, the reads and writes required by pg-jobs' public APIs, and sequence USAGE; it has no DDL, ownership, TRUNCATE, REFERENCES, TRIGGER or MAINTAIN rights. The internal jobs._sqlx_migrations table is never granted; reconciliation only removes any legacy access to it. Passing &[] is an immediate no-op, though ordinary applications need not call this API at all.

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`).
    // `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 handle = worker.start().await?;
    tokio::signal::ctrl_c().await?;
    handle.shutdown().await?;
    Ok(())
}

Worker is a single-use builder and Worker::start consumes it, returning the sole WorkerHandle; retain that handle for the runtime's lifetime. snapshot() is the current low-cardinality state, subscribe_snapshots() follows coalesced state changes, and subscribe_events() is a lossy, non-replaying operational stream whose receiver must handle broadcast::error::RecvError::Lagged.

request_shutdown() only signals. Consuming the handle with shutdown().await signals and then joins the runtime; join().await only waits for an exit that was requested elsewhere or caused by a supervised failure. Dropping the handle requests shutdown but cannot wait for it, so applications should explicitly call shutdown(). A graceful stop ceases to initiate claims, executes any job returned by a claim already engaged, and waits for all in-flight jobs plus the final Basic completion flush. It has no implicit timeout.

For one-shot work, Worker::drain consumes the worker and returns a DrainReport; total_claimed() preserves the former dispatched-job count. It installs the same listener and follows work that becomes visible while its local frontier is active, then stops at local quiescence. It does not prove that another replica has no gated work, and it runs neither the reaper nor scheduler services.

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

Execution contract

Every async Job runs without a handler-wide database transaction. The runtime owns the claim, heartbeat and short fenced settlement transactions; application code never receives the worker pool or a worker connection. Completion is separate from handler effects, so delivery is at least once: make handlers idempotent and use stable idempotency keys or an application outbox for external effects. A synchronous, blocking handler uses JobSync instead (see Synchronous / blocking jobs).

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

ctx.spawn and ctx.fan_out commit their children in short runtime-owned transactions before returning. A parent may therefore remain active while the same worker claims its children with spare local capacity, or another worker does so, subject to queue/group DB gates. A same-worker waiting parent needs more than one usable execution slot.

Shutdown stops this runtime from initiating further claims but still waits for its current parents. If a parent is waiting for children not yet claimed here, another active worker must advance them or shutdown waits until the parent returns. The same capacity caveat applies to finite drain().

impl Job for ScrapeChapter {
    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). Its handler can likewise run again within the configured business/infrastructure retry bounds, and it 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)
    }
}

Transaction-abort retries

The worker recognizes PostgreSQL deadlock and serialization aborts (40P01 and 40001) with one initial settlement attempt plus three retries after 10, 20 and 40 ms. Only the short terminal transaction is retried: a successful async, sync or Basic handler is never replayed by the local settlement loop. Non-flow Basic success batches retry the whole batch; fail, reschedule and DelayedGroup outcomes retry their complete short lifecycle operation.

When settlement exhausts its in-process budget, jobs_api_v1.release_transaction_retry puts an owned job back in pending after 100 ms without consuming its business attempt. SQL itself allows at most two such refunds for one attempt; the next exhaustion goes through normal retryable failure, restoring max_attempts and business backoff. If the row is no longer owned, fencing wins and the stale settlement cannot mutate it. Handler effects, including progress/log writes and application database writes, must remain idempotent because a bounded release or process crash can lead to a later claim. Direct users of the SQL lifecycle API must apply the same bounded whole-transaction retry contract themselves and use READ COMMITTED.

Writing to the database from a handler

Ctx deliberately exposes neither PgPool, PgConnection nor a general transaction helper. Inject an application repository or RLS-scoped pool into the job value when it is registered, and keep that credential distinct from the worker's pg-jobs role:

#[derive(JobKind)]
struct ReindexDocument {
    documents: DocumentRepository,
}

#[async_trait::async_trait]
impl Job for ReindexDocument {
    type Input = DocumentId;
    type Output = ();

    async fn run(&self, id: DocumentId, _ctx: &mut Ctx) -> Result<(), JobError> {
        self.documents.reindex_idempotently(id).await?;
        Ok(())
    }
}

registry.register(ReindexDocument { documents });

Use an idempotency key or an application outbox when database state and an external effect must move together. Low-level callers that already own an application transaction may use JobStore::complete_in_transaction; it is intentionally not available through Ctx.

Concurrency and connections

WorkerConfig::max_in_flight bounds handlers executing application code across Rich and Basic queues. Completed handlers hand their owned outcome to a separately bounded settlement stage, allowing Basic to form configured completion batches without consuming execution slots. The runtime uses worker connections only for short claim, capability, heartbeat and settlement operations; no handler pins one for its whole duration.

WorkerConfig::settlement_capacity overrides the second bound. Its default is 2 * max_in_flight without Basic queues, or basic_batch_size + 2 * max_in_flight when Basic is enabled. Capacity is reserved before an execution slot is released, keeping local ownership bounded by execution plus settlement. The worker snapshot exposes both capacity windows, per-pipeline executing, handoff_waiting and settling, total/high-water ownership, and settlement rows/flush reason/duration/fencing/recovery telemetry.

A connection pooler like PgBouncer caps total Postgres backends across many worker processes. 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. The Rust implementation is the workspace rooted at rust/Cargo.toml and is locked by rust/Cargo.lock; benchmark-manager has a separate workspace and lockfile. From the repository root, run:

docker compose up -d
export DATABASE_URL="postgres://postgres:postgres@localhost:5432/pg_jobs"
cargo test --manifest-path rust/Cargo.toml --workspace --all-targets --locked

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 --manifest-path rust/Cargo.toml --locked / cargo check --manifest-path rust/Cargo.toml --locked); only the tests need one.

Benchmark harness

The benchmark workspace crate is a reusable performance harness: twenty-four scenarios exercise the engine end to end (DAG flows, fan-out and parent/child frontier liveness, retry/DLQ, deferrals, concurrency caps, rate limits, pause/resume, the scheduler, GC/rollup, lifecycle contention), 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. The non-publishable benchmark-runner application is its Rust scenario executable. The root benchmark-manager supervises that hash-pinned external process through a typed exact-ID protocol and owns campaign evidence without linking pg-jobs.

cargo run --manifest-path rust/Cargo.toml --locked -p benchmark-runner -- execution catalog

See the benchmark guide for scenarios and result contracts, and the benchmark-manager guide for the complete proof-backed operator workflow.

License

ID	Version
anyhow	^1
async-trait	^0.1
chrono	^0.4
croner	^2
futures-util	^0.3
pg-jobs-derive	=0.10.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-21 00:48:39 +02:00

MIT

308 KiB

Assets (1)

pg-jobs-0.10.0.crate 308 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

pg-jobs (0.10.0)

Installation

About this package

pg-jobs

Features

Install

Migrations

Optional restricted runtime role

Usage

Execution contract

Synchronous / blocking jobs (`JobSync`)

Transaction-abort retries

Writing to the database from a handler

Concurrency and connections

Development

Benchmark harness

License

Dependencies

pg-jobs (0.10.0)

Installation

About this package

pg-jobs

Features

Install

Migrations

Optional restricted runtime role

Usage

Execution contract

Synchronous / blocking jobs (JobSync)

Transaction-abort retries

Writing to the database from a handler

Concurrency and connections

Development

Benchmark harness

License

Dependencies

Synchronous / blocking jobs (`JobSync`)