Crate job

Expand description

job is an async, Postgres-backed job scheduler and runner for Rust applications. It coordinates distributed workers, tracks job history, and handles retries with predictable backoff. Inspired by earlier systems like Sidekiq, it focuses on running your application code asynchronously, outside of request/response paths while keeping business logic in familiar Rust async functions. The crate uses sqlx for database access and forbids unsafe.

§Documentation

§Highlights

Durable Postgres-backed storage so jobs survive restarts and crashes.
Automatic exponential backoff with jitter, plus opt-in infinite retries.
Concurrency controls that let many worker instances share the workload, configurable through JobPollerConfig, plus a per-type concurrency cap via JobInitializer::max_concurrent_per_process.
Two singleton flavors beyond the default: keyed jobs — at most one LIVE job per (job_type, key), respawnable once terminal — via KeyedJobSpawner, and resident jobs — at most one job per type, ever, that never terminates — via ResidentJobSpawner.
Built-in migrations that you can run automatically or embed into your own migration workflow.

§Core Concepts

Jobs service – Jobs owns registration, polling, and shutdown.
Initializer – JobInitializer registers a job type and builds a JobRunner for each execution. Defines the associated Config type.
Spawner – JobSpawner is returned from registration and provides type-safe job creation methods. Parameterized by the config type.
Runner – JobRunner performs the work using the provided CurrentJob context.
Current job – CurrentJob exposes attempt counts, execution state, and access to the Postgres pool during a run.
Completion – JobCompletion returns the outcome: finish, retry, or reschedule at a later time.

§Lifecycle

Initialize the service with Jobs::init
Register initializers with Jobs::add_initializer – returns a JobSpawner
Start polling with Jobs::start_poll
Use spawners to create jobs throughout your application
Shut down gracefully with Jobs::shutdown

§Example

use async_trait::async_trait;
use job::{
    CurrentJob, Job, JobCompletion, JobId, JobInitializer, JobRunner,
    JobSpawner, JobSvcConfig, JobType, Jobs,
};
use serde::{Deserialize, Serialize};

// 1. Define your config (serialized to the database)
#[derive(Debug, Serialize, Deserialize)]
struct MyConfig {
    value: i32,
}

// 2. Define your initializer
struct MyInitializer;

impl JobInitializer for MyInitializer {
    type Config = MyConfig;

    fn job_type(&self) -> JobType {
        JobType::new("my-job")
    }

    fn init(&self, job: &Job) -> Result<Box<dyn JobRunner>, Box<dyn std::error::Error>> {
        let config: MyConfig = job.config()?;
        Ok(Box::new(MyRunner { value: config.value }))
    }
}

// 3. Define your runner
struct MyRunner {
    value: i32,
}

#[async_trait]
impl JobRunner for MyRunner {
    async fn run(
        &self,
        _current_job: CurrentJob,
    ) -> Result<JobCompletion, Box<dyn std::error::Error>> {
        println!("Processing value: {}", self.value);
        Ok(JobCompletion::Complete)
    }
}

// 4. Wire it up
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = JobSvcConfig::builder()
        .pg_con("postgres://user:pass@localhost/db")
        .build()?;

    let mut jobs = Jobs::init(config).await?;

    // Registration returns a type-safe spawner
    let spawner: JobSpawner<MyConfig> = jobs.add_initializer(MyInitializer);

    jobs.start_poll().await?;

    // Use the spawner to create jobs
    spawner.spawn(JobId::new(), MyConfig { value: 42 }).await?;

    Ok(())
}

§Scheduling

Jobs run immediately once a poller claims them. If you need a future start time, schedule it up front with JobSpawner::spawn_at_in_op. After a run completes, return JobCompletion::Complete for one-off work or use the JobCompletion::Reschedule* variants to book the next execution.

§Retries

Retry behaviour comes from JobInitializer::retry_on_error_settings. Once attempts are exhausted the job is marked as errored and removed from the queue.

impl JobInitializer for MyInitializer {
    // ...

    fn retry_on_error_settings(&self) -> RetrySettings {
        RetrySettings {
            n_attempts: Some(5),
            min_backoff: Duration::from_secs(10),
            max_backoff: Duration::from_secs(300),
            ..Default::default()
        }
    }
}

§Batched execution

Job types whose work is dominated by per-job transaction overhead — the common “an event spawns a command job that mutates one entity” shape — can opt into batched execution with Jobs::add_batched_initializer. The poller then hands a BatchedJobRunner every job of that type it claimed in one poll, so K jobs cost one transaction and one commit instead of K.

impl BatchedJobInitializer for RevalueInitializer {
    type Config = RevalueConfig;

    fn job_type(&self) -> JobType { JobType::new("command.revalue") }
    fn max_batch_size(&self) -> usize { 25 }

    fn init(&self, _: JobSpawner<Self::Config>)
        -> Result<Box<dyn BatchedJobRunner<Config = Self::Config>>, Box<dyn Error>>
    {
        Ok(Box::new(RevalueRunner { accounts: self.accounts.clone() }))
    }
}

#[async_trait]
impl BatchedJobRunner for RevalueRunner {
    type Config = RevalueConfig;

    async fn run_batch(&self, batch: CurrentBatchedJob<RevalueConfig>)
        -> Result<JobBatchCompletion, Box<dyn Error>>
    {
        let mut op = batch.begin_op().await?;
        for item in batch.items() {
            self.accounts.revalue_in_op(&mut op, item.config().account_id).await?;
        }
        Ok(JobBatchCompletion::CompleteAllWithOp(op))
    }
}

Spawning and awaiting are unchanged. Batched types use the same JobSpawner and resolve through the same await_completion — whether a job ran in a batch is invisible from both sides.

Points worth knowing before opting in:

At most one job per queue_id is ever in the same batch, because the poll query already claims at most one row per queue. Items arrive sorted by queue_id so concurrent batches take domain locks in a consistent order.
A batch of one is normal. Under light load batches are size 1; write run_batch to be correct at any length.
Per-job outcomes are available via JobBatchCompletion::WithOutcomes: complete some, reschedule others, fail the rest, all in one commit. Every job must get exactly one outcome — the dispatcher rejects a partial set rather than guessing.
Returning Err fails the whole batch, retrying each job under the type’s RetrySettings. Jobs on a second or later attempt are never batched, so a persistently failing job ends up retrying alone.
A running batch costs one unit of max_jobs_per_process, not one per job — a batch is one task, one transaction, one connection.
Claims are throttled by free batch slots. A type may have max_concurrent_per_process (default 2) batches running per process, and the poll query claims at most max_batch_size × free slots rows for it. Rows are therefore only locked when a batch is free to start on them: the rest of a backlog stays pending — claimable by other poller instances, and accumulating into fuller later batches. Raise the slot count to trade locked rows for more concurrency on a hot type.
Keep external calls out of batched runners. A shared transaction held across HTTP or mail delivery is a liability; leave those types on JobRunner.

§Per-type concurrency limits

JobInitializer::max_concurrent_per_process bounds how many jobs of one type may execute concurrently in THIS process; while every slot is busy the poller stops claiming rows of that type and the backlog stays pending — cheap database rows, visible to other poller instances — instead of pinning executor slots, connections, and memory. This is the direct defense against a slow external dependency (a hanging HTTP call, say) confiscating JobPollerConfig::max_jobs_per_process from every other job type.

impl JobInitializer for KeycloakSyncInitializer {
    // ...
    fn max_concurrent_per_process(&self) -> Option<usize> { Some(20) }
}

There is deliberately no cross-instance cap: the former max_concurrent_global made every poll pre-count the fleet’s running executions, for a bound that was only ever soft. max_concurrent_per_process is exact, free at the database, and multiplied by a known instance count gives a real fleet-wide ceiling. See PERFORMANCE.md for the measurements.

A capped type’s backlog is observable the same way any pending backlog is: via the crate’s existing stale-pending warnings (job.check_stale_pending_jobs), which fire regardless of why a type’s rows are sitting pending.

§Job flavors

Every job is one of three flavors — regular (what JobInitializer registers, optionally BatchedJobInitializer or spawned with a queue_id), keyed, or resident. “Flavor” is this crate’s umbrella term for the three whenever a doc comment or diagnostic needs to name them collectively.

flavor	admission invariant	terminates?	id
regular	none (`queue_id` = at most one RUNNING per queue at a time — a scheduling-time mutual exclusion, not a FIFO ordering guarantee)	yes	caller-chosen
keyed	at most one LIVE job per `(job_type, key)`; respawnable after terminal	yes	internal
resident	at most one job per `job_type`, EVER	cannot (no `Complete` variant exists)	internal

§Keyed jobs

For at-most-one-live-per-key semantics, register with Jobs::add_keyed_initializer and spawn with KeyedJobSpawner::spawn: at most one LIVE (pending/running) job of (job_type, key) at a time. A key becomes respawnable once its job reaches a terminal state — the next spawn creates a new generation under the same key. Enumerate every key of a type — the entry point for a “have my listeners caught up?” check — with Jobs::keyed_handles.

let shard_spawner = jobs.add_keyed_initializer(ShardListenerInitializer);

// One job per shard; re-spawning a live shard is a no-op that returns
// the existing job's handle. Once a shard's job goes terminal, spawning
// it again starts a new generation.
shard_spawner.spawn(format!("shard-{shard_id}"), ShardConfig { shard_id }).await?;

KeyedJobSpawner::spawn_in_op does the same inside a transaction you already have open, and KeyedJobSpawner::spawn_all / KeyedJobSpawner::spawn_all_in_op do it for many keys at once. Both report per key whether they created the job or resolved to one that already held the key (KeyedSpawn::created) — the collision is never silently dropped, unlike JobSpec::dedup_key on the regular spawn path:

let spawned = shard_spawner
    .spawn_all_in_op(&mut op, shards.iter().map(|s| {
        KeyedJobSpec::new(format!("shard-{s}"), ShardConfig { shard_id: *s })
    }).collect())
    .await?;
for s in &spawned {
    if s.created {
        // ... first-time setup for this shard, in the same transaction
    }
}

A spawn that resolves to a live holder changes nothing by default, which means a generation parked in the future (one that returned JobCompletion::RescheduleAt) waits for its own deadline no matter what arrives meanwhile. KeyedJobSpec::force_reschedule turns such a spawn into “run no later than the time I am asking for”: it pulls the holder’s execute_at forward to the spec’s own schedule_at (or now, when it has none), monotonically — earlier only, never later — and never over a retry backoff, reporting through KeyedSpawn::pulled_forward whether it moved anything.

// A held subscriber wakes as soon as work for its key shows up.
shard_spawner
    .spawn_all_in_op(&mut op, vec![
        KeyedJobSpec::new(format!("shard-{shard_id}"), cfg).force_reschedule(),
    ])
    .await?;

// ... or no later than a deadline of the caller's own, which pulls a
// longer hold in without dragging it all the way to now.
shard_spawner
    .spawn_all_in_op(&mut op, vec![
        KeyedJobSpec::new(format!("shard-{shard_id}"), cfg)
            .schedule_at(next_batch_window)
            .force_reschedule(),
    ])
    .await?;

By default a keyed generation’s execution state (see CurrentJob::update_execution_state) is deleted when it terminates, just like a regular job’s. Set KeyedJobInitializer::inherits_state to make it outlive the generation instead: it is kept (readable via JobSnapshot::execution_state), seeded into the next generation of that key, and older generations are compacted away at the next spawn — useful when a respawn should resume from a checkpoint rather than start cold.

§Resident jobs

For a process-wide singleton that just keeps running — a poller, a periodic sweep — register with Jobs::add_resident_initializer and spawn with ResidentJobSpawner::spawn: absolutely unique, at most one job of a type EVER exists. Once created it can never be spawned again — there is no respawn/new-generation escape hatch, unlike keyed jobs — and it can never complete either: ResidentJobRunner::run returns ResidentJobCompletion, a reschedule-only mirror of JobCompletion with no Complete variant, so a resident job finishing for good is a compile error, not a runtime surprise. Registration also forces eternal retry regardless of the initializer’s settings, for the same reason. spawn consumes the spawner, enforcing at the type level that only one job of this type can be created:

let cleanup_spawner = jobs.add_resident_initializer(CleanupInitializer);

// Consumes spawner - can't accidentally spawn twice
cleanup_spawner.spawn(CleanupConfig::default()).await?;

§Parameterized Job Types

For cases where the job type is configured at runtime (e.g., multi-tenant inboxes), store the job type in your initializer and return it from the instance method:

struct TenantJobInitializer {
    job_type: JobType,
    tenant_id: String,
}

impl JobInitializer for TenantJobInitializer {
    type Config = TenantJobConfig;

    fn job_type(&self) -> JobType {
        self.job_type.clone()  // From instance, not hardcoded
    }

    // ...
}

§Database migrations

See the setup guide for migration options and examples.

§Feature flags

es-entity enables advanced integration with the es_entity crate, allowing runners to finish with DbOp handles and enriching tracing/event metadata.

§Testing with simulated time

For deterministic testing of time-dependent behavior (e.g., backoff strategies), inject an artificial clock via JobSvcConfig::clock:

use job::{JobSvcConfig, ClockHandle};

let (clock, controller) = ClockHandle::manual();
let config = JobSvcConfig::builder()
    .pool(pool)
    .clock(clock)
    .build()?;

// Advance time deterministically
controller.advance(Duration::from_secs(60)).await;

Re-exports§

pub use error::JobError;

Modules§

error: Error type returned by the job service and helpers.

Structs§

BatchedJobItem: One job inside a batch: its identity, attempt, typed config and per-job execution state.
BulkSpawnResult: Return value of JobSpawner::spawn_all/JobSpawner::spawn_all_in_op.
Clock: Global clock access - like Utc::now() but testable.
ClockController: Controller for manual time operations.
ClockHandle: A handle to a clock for getting time and performing time-based operations.
CurrentBatchedJob: Context handed to a BatchedJobRunner for one batch.
CurrentJob: Context provided to a JobRunner while a job is executing.
Job: Entity capturing immutable job metadata and lifecycle events.
JobHandle: A minted, cloneable per-job capability: the public way to observe and await a job you did not run yourself.
JobHandles: An ordered collection of JobHandles.
JobId
JobOutcome: Outcome returned by JobHandle::await_completion, carrying both the terminal state and an optional return value.
JobPollerConfig: Controls how the background poller balances work across processes.
JobRegistry: Keeps track of registered job types and their retry behaviour.
JobReturnValue: Newtype wrapper around a raw JSON value representing the value produced by a job runner via CurrentJob::set_result.
JobSnapshot: A point-in-time view of a job, produced by JobHandle::load.
JobSpawner: A handle for spawning jobs of a specific type.
JobSpec: Describes a job to be created as part of a bulk JobSpawner::spawn_all call.
JobSvcConfig: Configuration consumed by Jobs::init. Build with JobSvcConfig::builder.
JobSvcConfigBuilder: Builder for JobSvcConfig.
JobType: Identifier describing a job type or class of work.
Jobs: Primary entry point for interacting with the Job crate. Provides APIs to register job handlers, manage configuration, and control scheduling and execution.
KeyedJobSpawner: A handle for spawning keyed jobs of a specific type.
KeyedJobSpec: Describes one keyed job to create as part of a bulk KeyedJobSpawner::spawn_all / KeyedJobSpawner::spawn_all_in_op call.
KeyedSpawn: The outcome of spawning one key.
ResidentJobSpawner: A handle for spawning the single resident job of a type.
RetrySettings: Controls retry attempt limits, telemetry escalation thresholds, and exponential backoff behaviour. Use RetrySettings::n_warn_attempts to decide how many failures remain WARN events before escalation. Set it to None to keep every retry at WARN.

Enums§

BatchItemOutcome: How a single job within a batch should be progressed.
BisectBudget: How many probes CurrentBatchedJob::run_bisected_with may spend on one batch. Re-exported from es-entity, whose BatchIsolation trait now owns the bisect search this budget governs — see its docs for the full contract (Auto‘s formula, MaxProbes’ clamping, FullResolution’s cost). How many probes a bisect may spend before giving up on the ranges it has not yet resolved.
JobBatchCompletion: Result returned by BatchedJobRunner::run_batch.
JobCompletion: Result returned by JobRunner::run describing how to progress the job.
JobEvent
JobStatus: Runtime status of a job, carried by JobSnapshot::state.
JobSvcConfigBuilderError: Error type for JobSvcConfigBuilder
JobTerminalState: Terminal outcome of a job lifecycle.
ResidentJobCompletion: Result returned by ResidentJobRunner::run describing how to progress a resident job. A reschedule-only mirror of JobCompletion — there is deliberately no Complete variant, so a resident job accidentally finishing is a compile error rather than a job that silently stops existing.

Constants§

DEFAULT_MAX_BATCH_SIZE: Default upper bound on the number of jobs handed to one run_batch call.
DEFAULT_MAX_CONCURRENT_BATCHES: Default number of batches of one type that may run concurrently per process.

Traits§

BatchedJobInitializer: Describes how to construct a BatchedJobRunner for a given job type.
BatchedJobRunner: Implemented by executors that process many jobs of one type together.
IncludeMigrations: Extend an sqlx::migrate!() call with Job’s migrations.
JobInitializer: Describes how to construct a JobRunner for a given job type.
JobOutcomes: Extension trait for inspecting a batch of JobOutcome values.
JobRunner: Implemented by job executors that perform the actual work.
KeyedJobInitializer: Describes how to construct a crate::JobRunner for a keyed job type. The keyed counterpart of crate::JobInitializer — keyed jobs use the ordinary crate::JobRunner/crate::JobCompletion (they legitimately complete; that’s what makes them respawnable), so only registration and spawning are distinct.
ResidentJobInitializer: Describes how to construct a ResidentJobRunner for a resident job type. The resident counterpart of crate::JobInitializer.
ResidentJobRunner: Implemented by resident job executors. Mirrors JobRunner except its result type cannot express completion.

Type Aliases§

BatchOutcomes: Per-job dispositions returned by a batched runner.

Crate job

Crate job Copy item path

§Documentation

§Highlights

§Core Concepts

§Lifecycle

§Example

§Scheduling

§Retries

§Batched execution

§Per-type concurrency limits

§Job flavors

§Keyed jobs

§Resident jobs

§Parameterized Job Types

§Database migrations

§Feature flags

§Testing with simulated time

Re-exports§

Modules§

Structs§

Enums§

Constants§

Traits§

Type Aliases§

Crate job