Cookbook / Data and storage

Background jobs

Durable jobs on PostgreSQL with retries, delays, unique keys and graceful shutdown.

Some work should not happen inside a request: sending webhooks, calling slow APIs, generating reports. axeon-jobs keeps jobs in a PostgreSQL table, so a job can be saved in the *same transaction* as the data it belongs to, with no extra infrastructure to run.

Terminal

cargo add axeon-jobs axeon-orm --features axeon-orm/postgres

src/main.rs

use axeon::{Json, Server, ServerError};
use axeon::json;
use axeon_jobs::{Enqueue, Job, JobContext, JobResult, Worker};
use axeon_orm::prelude::*;
use serde::{Deserialize, Serialize};
use std::time::Duration;

#[derive(Serialize, Deserialize)]
struct SendWebhook {
    url: String,
    event_id: i64,
}

impl Job for SendWebhook {
    const NAME: &'static str = "send_webhook";
    const QUEUE: &'static str = "webhooks";
    const MAX_ATTEMPTS: u32 = 8;                       // default 5
    const TIMEOUT: Duration = Duration::from_secs(30); // default 5 minutes

    async fn run(self, ctx: JobContext) -> JobResult {
        // Jobs run at least once: use the job's ID as an idempotency key.
        tracing::info!(job = %ctx.id(), url = %self.url, event = self.event_id, "sending webhook");
        // `?` on any error retries with backoff; JobError::fatal gives up.
        Ok(())
    }
}

/// Runs every night. Lower priority than webhooks, so it never delays them.
#[derive(Serialize, Deserialize)]
struct Cleanup {
    older_than_days: u32,
}

impl Job for Cleanup {
    const NAME: &'static str = "cleanup";
    const PRIORITY: i16 = -5; // higher numbers are taken first; the default is 0

    async fn run(self, _ctx: JobContext) -> JobResult {
        tracing::info!(days = self.older_than_days, "deleting old rows");
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tracing_subscriber::fmt().init();
    let db = Db::connect(&std::env::var("DATABASE_URL")?).await?;

    // Creates the jobs table. Or add `axeon_jobs::migration()` to your migrations.
    axeon_jobs::install(&db).await?;

    // The worker runs beside the web server (or in its own process).
    let worker = Worker::new(db.clone())
        .register::<SendWebhook>()
        .queue("webhooks", 16) // concurrency; default 4 per queue
        // A cron schedule (UTC). Every worker may declare it: each run is enqueued once.
        .schedule("nightly-cleanup", "0 3 * * *", Cleanup { older_than_days: 90 });
    tokio::spawn(async move {
        if let Err(error) = worker.run_until(axeon::shutdown_signal()).await {
            tracing::error!(%error, "worker stopped");
        }
    });

    let mut app = Server::new();
    app.state(db);

    app.post("/events/:id", |req| async move {
        let event_id: i64 = req.param("id")?;
        let db = req.db()?;
        let job = SendWebhook { url: "https://example.com/hook".into(), event_id };

        // Delayed, and skipped while an identical one is waiting or running.
        axeon_jobs::enqueue_with(
            db,
            &job,
            Enqueue::new().delay(Duration::from_secs(5)).unique_key(format!("event:{}", event_id)),
        )
        .await
        .map_err(|e| ServerError::InternalError(e.to_string()))?;
        Ok::<_, ServerError>((202, Json(json!({ "queued": event_id }))))
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    app.serve_with_shutdown(listener, axeon::shutdown_signal()).await?;
    Ok(())
}

Terminal

export DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres
cargo run
curl -X POST http://127.0.0.1:3000/events/7

How it behaves#

Workers claim due jobs with FOR UPDATE SKIP LOCKED, so any number of workers share the table and each job runs on one at a time. They wake on LISTEN/NOTIFY, sleep until the next delayed job is due, and poll every five seconds as a safety net.
A claimed job is *leased* for its timeout plus 30 seconds instead of holding a transaction open, so long jobs do not pin a database connection. If a worker dies, another takes the job when the lease ends.
Success deletes the job. An error, panic or timeout retries with backoff (2s, 4s, 8s, up to an hour, with jitter). Out of attempts, or after JobError::fatal, the job is marked failed and kept for inspection.
Workers only take job names they have registered, so old and new versions can run side by side during a deploy.
On shutdown a worker stops taking jobs and waits for running ones. Any still running are cancelled and put back without using an attempt.

Notes#

Enqueue with &mut tx instead of &db to commit the job together with your other writes. The job exists only if the transaction commits.
Worker::metrics(&metrics) reports completed and failed jobs, durations and how many are running on your Prometheus endpoint.
Jobs::new(db) gives counts per queue and state, plus retry, cancel and prune_failed, for an operator screen.
Cron schedules (Worker::schedule) use five fields in UTC, with names, ranges, steps and @daily-style shortcuts. The next run time lives in a table and is advanced by exactly one worker, so declaring the same schedule on every instance is fine. A run is skipped while the previous one is still waiting or running, and runs missed while no worker was up are made up once, not once per missed run.
Priorities: among due jobs on a queue, higher PRIORITY values go first. Set one per job type or per enqueue with Enqueue::priority.
Upgrading from axeon-jobs 0.1: run axeon_jobs::install(&db) once, or add axeon_jobs::migrations() to your migrations, before starting 0.2 workers. The schema gained a priority column and a schedules table.