Rustango docs
← Anleitungen

Hintergrund-Jobs

Manche Arbeit sollte nicht während eines Requests passieren — eine Willkommens-E-Mail senden, einen Upload skalieren, eine Drittanbieter-API synchronisieren. Es inline zu tun lässt den Benutzer warten und koppelt die Response an einen unzuverlässigen externen Aufruf. Ein Hintergrund-Job verschiebt diese Arbeit auf eine Queue: der Handler kehrt sofort zurück, und ein Pool von Workern führt den Job Augenblicke später aus, mit automatischen Retries und einem Dead-Letter-Pfad für Fehlschläge. Das ist Django-Q / Celery / Laravel-Queues, in Rust.

Background jobs in Rustango: a handler dispatches a Job onto a queue, worker tasks run it, retryable failures back off and retry, fatal ones go to a dead-letter handler

Ein Begriff hier neu für dich? Queue, Worker, Retry/Backoff, Dead-Letter — siehe das Glossar.

Quelle: rustango::jobs (Job, JobQueue, InMemoryJobQueue, JobError, JobDeadLetter) und rustango::jobs::DatabaseJobQueue (die tabellengestützte Queue, Alias von pg::PgJobQueue) — hinter der jobs-Feature (standardmäßig aktiv).

Ausführbare Version: die In-Memory-, Retry- und Dead-Letter-Snippets sind aus jobs_doc.rs kopiert (cargo test -p rustango --test jobs_doc); die persistente Queue wird per Dogfooding auf SQLite durch jobs_sqlite_live.rs erprobt (cargo test -p rustango --features sqlite,jobs-postgres --test jobs_sqlite_live).

Inhaltsverzeichnis

Schritt 1 — Einen Job definieren

Ein Job ist ein serialisierbares Struct, das Job implementiert. Die Payload ist das, was in die Queue gestellt wird (nach JSON serialisiert); run() ist die Arbeit. NAME routet eine in die Queue gestellte Payload zurück zu ihrem Handler, er muss also eindeutig sein.

Erzeuge ein Gerüst mit der CLI — cargo run -- make:job WelcomeEmail — oder schreibe es von Hand:

use rustango::jobs::{Job, JobError};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct WelcomeEmail {
    user_id: i64,
}

#[async_trait::async_trait]   // add `async-trait` to your Cargo.toml
impl Job for WelcomeEmail {
    const NAME: &'static str = "welcome_email";

    async fn run(&self) -> Result<(), JobError> {
        // ... look up the user and send the email
        Ok(())
    }
}

run() gibt zurück:

  • Ok(()) — erledigt.
  • Err(JobError::Retryable(msg)) — transient; der Worker wiederholt mit Backoff.
  • Err(JobError::Fatal(msg)) — permanent; überspringe Retries, Dead-Letter es sofort.

Überschreibe const MAX_ATTEMPTS: u32 = 3; auf dem Impl, um die Retry-Obergrenze zu ändern (Standard 5).

Schritt 2 — Eine Queue starten

Für einen einzelnen Prozess (und für Dev und Tests) führt InMemoryJobQueue Jobs auf tokio-Worker-Tasks aus — ohne Datenbank. Registriere jeden Job-Typ, dann start() die Worker (Jobs werden nicht aufgenommen, bis du es tust):

use rustango::jobs::{InMemoryJobQueue, JobQueue};
use std::sync::Arc;

let queue = Arc::new(InMemoryJobQueue::with_workers(4));   // 4 worker tasks
queue.register::<WelcomeEmail>().await;                    // before start/dispatch
queue.start().await;

// ... app runs ...

queue.shutdown().await;   // on shutdown: drain in-flight jobs, then stop

Behalte den Arc<InMemoryJobQueue> in deinem App-State, damit Handler ihn erreichen können.

In-Memory bedeutet In-Memory. In der Queue stehende oder in Bearbeitung befindliche Jobs gehen beim Neustart verloren. Für alles, dessen Verlust du dir nicht leisten kannst, verwende die persistente Queue.

Schritt 3 — Aus einem Handler dispatchen

Sobald die Queue läuft, ist das Dispatchen ein einziger Aufruf — er stellt in die Queue und kehrt sofort zurück, sodass der Request nicht auf die Arbeit wartet:

queue.dispatch(&WelcomeEmail { user_id: 42 }).await?;

Ein Worker nimmt ihn auf und führt run() Augenblicke später aus. Durchgängig verifiziert: alle drei dispatchten Jobs laufen auf den Workern.

for user_id in 1..=3 {
    queue.dispatch(&WelcomeEmail { user_id }).await.unwrap();
}
// → all three WelcomeEmail::run() calls execute on the worker pool

Schritt 4 — In deine App verdrahten

Die Schritte 1–3 zusammengesetzt: baue die Queue und registriere jeden Job einmalig beim Boot, starte die Worker, übergib die Queue an deine Routen, damit Handler sie erreichen können, und drain dann beim Shutdown. Die Queue lebt in deiner main.rs:

// src/main.rs
use std::sync::Arc;
use rustango::jobs::{InMemoryJobQueue, JobQueue};
use rustango::manage::Cli;

#[rustango::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Build the queue + register every job type BEFORE starting workers.
    let queue = Arc::new(InMemoryJobQueue::with_workers(4));
    queue.register::<WelcomeEmail>().await;
    queue
        .on_dead_letter(|dl| async move {
            tracing::error!(job = dl.name, attempts = dl.attempts, error = %dl.error,
                            "job dead-lettered");
        })
        .await;

    // 2. Start the workers — tokio tasks that run inside THIS process.
    queue.start().await;

    // 3. Share the queue with handlers (axum extension/state).
    let app = urls::api().layer(axum::Extension(queue.clone()));

    // 4. Boot the server. Cli::run() blocks until Ctrl-C / SIGTERM.
    Cli::new().api(app).with_health().run().await?;

    // 5. On shutdown: drain in-flight jobs, then stop.
    queue.shutdown().await;
    Ok(())
}

Ein Handler dispatcht, indem er die Queue aus dem Request wieder ausliest:

use axum::Extension;
use std::sync::Arc;
use rustango::jobs::{InMemoryJobQueue, JobQueue};

async fn signup(Extension(queue): Extension<Arc<InMemoryJobQueue>>) -> StatusCode {
    // ... create the user ...
    queue.dispatch(&WelcomeEmail { user_id: 42 }).await.ok();  // returns instantly
    StatusCode::CREATED
}

Speichere den konkreten Queue-Typ. Der JobQueue-Trait ist nicht object-safe (seine register/dispatch sind generisch), du kannst also nicht Arc<dyn JobQueue> im State halten — behalte den konkreten Arc<InMemoryJobQueue> (oder Arc<DatabaseJobQueue>).

Die Worker laufen lassen (CLI + Produktion)

start() spawnt die Worker als tokio-Tasks innerhalb des aktuellen Prozesses. Wenn du also deine App mit der CLI startest — cargo run (das den Server ausführt) — laufen die Worker direkt daneben. Für die meisten Apps ist das alles, was du brauchst: ein Prozess bedient Requests und leert die Queue; es gibt keinen separaten Worker-Befehl.

cargo run                 # starts the server AND the in-process workers
cargo run -- make:job WelcomeEmail   # scaffold a new job type

Ein dedizierter Worker-Prozess (Produktion)

Im großen Maßstab willst du die Worker oft getrennt von der Web-Ebene — damit ein Traffic-Spike keine Jobs aushungern kann und du beide unabhängig skalierst. Mit der persistenten Queue zieht jeder Prozess aus derselben rustango_jobs-Tabelle, führe also einfach ein zweites, serverloses Binary aus, das die Queue baut, startet und bis zu einem Signal blockiert:

// src/bin/worker.rs — run with `cargo run --bin worker`
use std::sync::Arc;
use rustango::jobs::{DatabaseJobQueue, JobQueue};

#[rustango::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = /* connect your tri-dialect Pool */;
    DatabaseJobQueue::ensure_table_pool(&pool).await?;

    let queue = Arc::new(DatabaseJobQueue::with_workers_pool(pool, 8));
    queue.register::<WelcomeEmail>().await;   // register the SAME job types
    queue.start().await;

    tokio::signal::ctrl_c().await?;            // block until Ctrl-C / SIGTERM
    queue.shutdown().await;                     // drain in-flight, then exit
    Ok(())
}

Deploye es als eigenen Container/Service und skaliere auf N Replicas — sie ziehen alle sicher aus der geteilten Tabelle. Der Web-Prozess muss dann nur noch dispatch (er muss keine Worker start()). Kopple den Worker mit einem periodischen reclaim_stuck_jobs_pool-Sweep, um Jobs eines abgestürzten Workers zurückzuholen.

Retries und Backoff

Ein Job, der Retryable zurückgibt, wird mit exponentiellem Backoff (1s, 2s, 4s, 8s, …) bis zu MAX_ATTEMPTS erneut in die Queue gestellt. Verwende es für transiente Fehler — einen Timeout, eine rate-limitierte API, einen Deadlock:

#[async_trait::async_trait]
impl Job for FlakyImport {
    const NAME: &'static str = "flaky_import";
    async fn run(&self) -> Result<(), JobError> {
        match call_external_api().await {
            Ok(_)  => Ok(()),
            Err(e) => Err(JobError::Retryable(e.to_string())),   // try again later
        }
    }
}

Der zugrunde liegende Test dispatcht einen Job, der einmal fehlschlägt und dann gelingt, und stellt sicher, dass er mehr als einmal lief und schließlich gelang — das Retry findet tatsächlich statt.

Der Dead-Letter-Handler

Wenn ein Job seine Retries erschöpft — oder sofort Fatal zurückgibt — wird er an den Dead-Letter-Callback übergeben, statt zu verschwinden. Registriere einen (vor start), um den Fehlschlag zu loggen, zu alarmieren oder zu persistieren:

queue
    .on_dead_letter(|dl| async move {
        // dl: JobDeadLetter { name, payload, attempts, error }
        tracing::error!(job = dl.name, attempts = dl.attempts, error = %dl.error,
                        "job dead-lettered");
    })
    .await;

Ein Fatal-Fehler überspringt Retries und landet hier beim ersten Versuch:

async fn run(&self) -> Result<(), JobError> {
    Err(JobError::Fatal("unprocessable payload".into()))   // → dead-letter now
}

Der Test bestätigt, dass der Callback für einen Fatal-Job genau einmal feuert, mit intaktem name und error des Jobs.

Die persistente Queue (Produktion)

InMemoryJobQueue ist Einzelprozess und vergisst beim Neustart. Für echte Deployments — mehrere Worker, mehrere Replicas, Jobs, die einen Absturz überstehen müssen — verwende DatabaseJobQueue (derselbe Typ wie PgJobQueue). Jobs leben in einer rustango_jobs-Tabelle; Worker nehmen sie mit einem transaktionsbegrenzten UPDATE … RETURNING auf, es ist also über Prozesse hinweg sicher. Sie ist tri-dialektisch (PostgreSQL, MySQL, SQLite). Die Job-Definitionen aus Schritt 1 bleiben unverändert.

use rustango::jobs::DatabaseJobQueue;   // = rustango::jobs::pg::PgJobQueue
use rustango::jobs::JobQueue;
use std::sync::Arc;
use std::time::Duration;

// 1. Create the jobs table once at boot (idempotent, tri-dialect).
DatabaseJobQueue::ensure_table_pool(&pool).await?;

// 2. Build the queue over your pool + N workers, and start it.
let queue = Arc::new(
    DatabaseJobQueue::with_workers_pool(pool.clone(), 4)
        .poll_interval(Duration::from_millis(50)),
);
queue.register::<WelcomeEmail>().await;
queue.start().await;

// 3. Dispatch exactly as before — now it's durable.
queue.dispatch(&WelcomeEmail { user_id: 42 }).await?;

Abgestürzte Worker wiederherstellen. Stirbt ein Worker mitten im Job, bleibt die Zeile gesperrt. Führe einen periodischen Sweep aus, um Sperren älter als eine Schwelle freizugeben, damit ein anderer Worker den Job wieder aufnimmt:

// e.g. from a scheduled task every few minutes:
let reclaimed = DatabaseJobQueue::reclaim_stuck_jobs_pool(&pool, Duration::from_secs(300)).await?;

Sowohl der Dispatch-and-Run-Fluss als auch reclaim_stuck_jobs_pool werden per Dogfooding gegen SQLite in jobs_sqlite_live.rs erprobt.

Referenz

Backends

BackendVerwenden fürÜbersteht Neustart?
InMemoryJobQueueEinzelprozess, Dev, Testsnein
DatabaseJobQueue (PgJobQueue)Produktion; Multi-Worker / Multi-Replicaja (rustango_jobs-Tabelle)

JobError

VarianteWirkung
Retryable(String)erneut in die Queue mit exponentiellem Backoff, bis zu MAX_ATTEMPTS
Fatal(String)Retries überspringen → sofort Dead-Letter
Queue(String)interner Queue-Fehler (Serialisierung/Registrierung)

JobQueue-Methoden: register::<T>() · dispatch(&payload) · start() · shutdown() · pending_count(). DatabaseJobQueue ergänzt ensure_table_pool, with_workers_pool, poll_interval und reclaim_stuck_jobs_pool.

Siehe auch

  • Scheduler — für zeitbasierte wiederkehrende Arbeit (cron-artig), im Gegensatz zu On-Demand-Jobs.
  • E-Mail — die kanonische „mach es in einem Job"-Workload.
  • Caching — der andere Weg, Request-Handler schnell zu halten.
  • Signals — Fire-and-Forget-Hooks, die oft einen Job dispatchen.
← Anleitungen