Job

An async / distributed job runner for Rust applications with Postgres backend.

Uses sqlx for interfacing with the DB.

Features

• Async job execution with PostgreSQL backend
• Job scheduling and rescheduling
• Configurable retry logic with exponential backoff
• Built-in job tracking and monitoring

Usage

Add this to your Cargo.toml:

[dependencies]
job = "0.1"

Basic Example

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use job::*;

// Define your job configuration
#[derive(Debug, Serialize, Deserialize)]
struct MyJobConfig {
    delay_ms: u64,
}
impl JobConfig for MyJobConfig {
    type Initializer = MyJobInitializer;
}

struct MyJobInitializer;
impl JobInitializer for MyJobInitializer {
    fn job_type() -> JobType {
        JobType::new("my_job")
    }

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

struct MyJobRunner {
    config: MyJobConfig,
}

#[async_trait]
impl JobRunner for MyJobRunner {
    async fn run(
        &self,
        _current_job: CurrentJob,
    ) -> Result<JobCompletion, Box<dyn std::error::Error>> {
        // Simulate some work
        tokio::time::sleep(tokio::time::Duration::from_millis(self.config.delay_ms)).await;
        println!("Job completed!");
        
        Ok(JobCompletion::Complete)
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Connect to database (requires PostgreSQL with migrations applied)
    let pool = sqlx::PgPool::connect("postgresql://user:pass@localhost/db").await?;
    
    // Create Jobs service
    let config = JobsSvcConfig::builder()
        .pg_con("postgresql://user:pass@localhost/db")
        // If you are using sqlx and already have a pool
        // .pool(sqlx::PgPool::connect("postgresql://user:pass@localhost/db")
        .build()
        .expect("Could not build JobSvcConfig");
    let mut jobs = Jobs::init(config).await?;
    
    // Register job type
    jobs.add_initializer(MyJobInitializer);
    
    // Start job processing
    // Must be called after all initializers have been added
    jobs.start_poll().await?;
    
    // Create and spawn a job
    let job_config = MyJobConfig { delay_ms: 1000 };
    let job_id = JobId::new();
    let job = jobs.create_and_spawn(job_id, job_config).await?;
    
    // Do some other stuff...
    tokio::time::sleep(tokio::time::Duration::from_millis(5000)).await;

    // Check if its completed
    let job = jobs.find(job_id).await?;
    assert!(job.completed());
    
    Ok(())
}

Setup

In order to use the jobs crate migrations need to run on Postgres to initialize the tables. You can either let the library run them, copy them into your project or add them to your migrations via code.

Option 1. Let the library run the migrations - this is useful when you are not using sqlx in the rest of your project. To avoid compilation errors set export SQLX_OFFLINE=true in your dev shell.

let config = JobsSvcConfig::builder()
    .pool(sqlx::PgPool::connect("postgresql://user:pass@localhost/db")
    // set to true by default when passing .pg_con("<con>") - false otherwise
    .exec_migration(true)
    .build()
    .expect("Could not build JobSvcConfig");

Option 2. If you are using sqlx you can copy the migration file into your project:

cp ./migrations/20250904065521_job_setup.sql <path>/<to>/<your>/<project>/migrations/

Option 3. You can also add the job migrations in code when you run your own migrations without copying the file:

use job::IncludeMigrations;
sqlx::migrate!().include_job_migrations().run(&pool).await?;

Optional Features

tokio-task-names

Enables named tokio tasks for better debugging and observability. This feature requires both the feature flag AND setting the tokio_unstable compiler flag.

To enable this feature:

[dependencies]
job = { version = "0.1", features = ["tokio-task-names"] }

And in your .cargo/config.toml:

[build]
rustflags = ["--cfg", "tokio_unstable"]

Important: Both the feature flag AND the tokio_unstable cfg must be set. The feature alone will not enable task names - it requires the unstable tokio API which is only available with the compiler flag.

When fully enabled, all spawned tasks will have descriptive names like job-poller-main-loop, job-{type}-{id}, etc., which can be viewed in tokio-console and other diagnostic tools.

Telemetry

job emits structured telemetry via tracing spans such as job.poll_jobs, job.fail_job, and job.complete_job. Failed attempts push fields like error, error.message, error.level, and will_retry, so you can stream the events into your existing observability pipeline without wrapping the runner in additional logging.

The RetrySettings that you configure for each JobInitializer directly influence that telemetry:

• n_attempts caps how many times the dispatcher will retry before emitting a terminal ERROR and deleting the job execution.
• n_warn_attempts controls how many consecutive failures remain WARN level events before the crate promotes them to ERROR. Setting it to None keeps every retry at WARN.
• min_backoff, max_backoff, and backoff_jitter_pct determine the delay that is recorded in the job.fail_job span before the next retry is scheduled.
• attempt_reset_after_backoff_multiples lets a job be considered healthy again after enough idle time (measured as multiples of the last backoff); the dispatcher resets the reported attempt counter accordingly.

Together these make the emitted telemetry reflect both the severity and cadence of retryable failures, which is especially helpful when wiring the crate into alerting systems.

License

Licensed under the Apache License, Version 2.0.