Tasks plugin

umbral-tasks runs background work outside the request/response cycle. It uses your application's existing database as the broker - no Redis, no RabbitMQ, no separate infrastructure. A fresh umbral project gets background jobs for the cost of one .plugin(TasksPlugin) line.

Install

cargo add umbral-tasks

What you get from TasksPlugin::default()

Registering the plugin creates one migration: a task_row table that tracks every enqueued job - its name, JSON payload, status, attempt count, schedule time, and error message on failure. The plugin also contributes a tasks-worker management subcommand reachable via cargo run -- tasks-worker (your project's binary dispatches it through umbral_cli::dispatch(app)).

App::builder()
    .plugin(TasksPlugin::default())
    .build()?;

Declaring a task with #[task]

The recommended way to declare a task is with the #[task] attribute macro. Define a typed payload struct that implements serde::Deserialize, annotate the handler, and call the generated registration function at boot time.

use serde::{Deserialize, Serialize};
 
#[derive(Serialize, Deserialize)]
pub struct WelcomeEmailPayload {
    pub user_id: i64,
    pub locale: String,
}
 
#[umbral::task]
async fn send_welcome(payload: WelcomeEmailPayload) -> Result<(), String> {
    // ... send the email
    Ok(())
}

That's the whole wiring: the handler is discovered automatically. The macro registers the task in a link-time slice, and TasksPlugin (plus the tasks-worker / tasks-beat commands) installs every discovered handler at startup — there is nothing to call by hand, so a forgotten registration can no longer surface as a runtime HandlerNotFound in production.

The macro also emits, for when you want manual control:

1. The original async fn send_welcome unchanged, so you can call it directly in tests or other code.
2. A companion pub fn register_send_welcome() that installs the handler by hand — useful with TasksPlugin::no_discover(), which turns discovery off for the whole process and returns you to explicit registration.

Enqueue work anywhere in your request handlers:

use umbral_tasks::{EnqueueOptions, enqueue};
 
let _task_id = enqueue(
    "send_welcome",
    WelcomeEmailPayload { user_id: 42, locale: "en".to_string() },
    EnqueueOptions::default(),
).await?;

Overriding the task name

By default the task name equals the Rust function identifier. Override it with name = "..." when you need a dotted or namespaced key:

#[umbral::task(name = "email.send_welcome")]
async fn send_welcome(payload: WelcomeEmailPayload) -> Result<(), String> {
    Ok(())
}
 
// Discovery registers it under the custom name; enqueue uses it too:
enqueue("email.send_welcome", payload, EnqueueOptions::default()).await?;

Constraints the macro enforces

The macro emits a compile_error! with a targeted message if any of these rules are violated:

Enqueue by type, not by string

#[task] also generates a typed handle named after the function — send_welcome → SendWelcome. Enqueue through it and the task's name never appears at the call site:

#[task]
async fn send_welcome(payload: Welcome) -> Result<(), String> { /* ... */ }
 
// somewhere in a handler
SendWelcome::enqueue(Welcome { user_id: 7 }).await?;
SendWelcome::enqueue_with(payload, EnqueueOptions { max_attempts: Some(5), ..Default::default() }).await?;

enqueue("send_welcom", payload, opts).await?;   // typo — compiles fine

Scheduled tasks take the same treatment, and need it more — a beat entry pointing at a deleted task fires forever into HandlerNotFound and is the last place anyone looks:

TasksPlugin::default()
    .periodic_task::<SendMatchReminder>("reminders", Schedule::cron("0 * * * *"), payload)

umbral_tasks::enqueue_task::<T>(payload, opts) is the same thing when you're generic over the task. The string API (enqueue("name", …), .periodic(…, "name", …)) still works — this is additive, and dynamic registration (below) still needs it.

Dynamic registration (advanced)

register_handler is available directly when you need to register a closure dynamically - for example, when the handler name is determined at runtime or when you need to skip the macro for a generic handler.

use umbral_tasks::{EnqueueOptions, enqueue, register_handler};
use serde::{Deserialize, Serialize};
 
#[derive(Serialize, Deserialize)]
struct WelcomePayload { user_id: i64 }
 
// At startup - before the worker runs.
register_handler("send_welcome_email", |payload: &str| async move {
    let p: WelcomePayload = serde_json::from_str(payload).map_err(|e| e.to_string())?;
    send_email(p.user_id).await.map_err(|e| e.to_string())
});
 
// In a handler, after a user signs up.
let _task_id = enqueue(
    "send_welcome_email",
    WelcomePayload { user_id: 42 },
    EnqueueOptions::default(),
).await?;

The #[task] macro expands to exactly this pattern. Use it directly only when the macro's single-payload constraint does not fit your case.

Running the worker

# Run continuously - polls every second, handles Ctrl-C gracefully.
cargo run -- tasks-worker
 
# Run exactly one claim/dispatch iteration and exit (useful for tests or cron).
cargo run -- tasks-worker --once

The worker claims one pending row at a time inside a transaction, dispatches the handler, and writes back succeeded or failed. Panicking handlers are caught via tokio::task::spawn so one bad job cannot take the worker down.

On Postgres, the claim SELECT uses FOR UPDATE SKIP LOCKED, so many workers running against the same queue each grab a different pending row instead of all contending on (and blocking behind) the head row's lock — the standard way to scale a database-backed queue horizontally. On SQLite the clause is omitted (its single-writer model already serializes claimers); correctness on both backends still rests on the conditional UPDATE ... WHERE status = 'pending' that only counts a claim if it actually transitioned the row. The same QuerySet::for_update_skip_locked() primitive is available for your own queue-like tables.

EnqueueOptions lets you set max_attempts (default 3), scheduled_for (default Utc::now()), and the scheduling/timeout knobs below.

Delayed and scheduled execution (eta / delay)

Every task carries a run_at instant: the earliest moment the worker is allowed to claim it. The dequeue query only claims rows whose run_at <= now(), so a task with a future run_at stays invisible until its time comes. By default run_at is set to enqueue time, so tasks run as soon as a worker is free.

Two EnqueueOptions fields push a task into the future:

• delay: Option<Duration> - run after a relative delay (run_at = now + delay).
• eta: Option<DateTime<Utc>> - run at an absolute instant. eta takes precedence over delay if both are set.

use std::time::Duration;
use chrono::Utc;
use umbral_tasks::{EnqueueOptions, enqueue};
 
// Run in ~60 seconds.
enqueue("send_welcome", payload.clone(), EnqueueOptions {
    delay: Some(Duration::from_secs(60)),
    ..EnqueueOptions::default()
}).await?;
 
// Run at a specific time.
enqueue("send_welcome", payload, EnqueueOptions {
    eta: Some(Utc::now() + chrono::Duration::hours(2)),
    ..EnqueueOptions::default()
}).await?;

Priority

Each task carries a priority (an i32). Higher number = claimed first. The worker orders the queue by priority descending before anything else, so a high-priority task jumps ahead of normal work even if it was enqueued later. The default is 0 (normal); a positive value runs sooner, a negative value drains behind the defaults.

use umbral_tasks::{EnqueueOptions, enqueue};
 
// Urgent - claimed before any priority-0 work waiting in the queue.
enqueue("send_password_reset", payload, EnqueueOptions {
    priority: Some(9),
    ..EnqueueOptions::default()
}).await?;

FIFO within a priority. Ties on priority break by scheduled_for then id, so two tasks at the same priority are claimed in the order they were enqueued - priority changes which band runs first, never the fairness inside a band.

Exponential-backoff retries

When a handler fails retriably (returns Err, panics, or times out - anything other than a missing handler), the worker no longer re-queues it immediately. Instead it pushes run_at into the future by an exponential backoff:

run_at = now + min(retry_backoff_base * 2^(attempts - 1), retry_backoff_max)

So the first retry waits retry_backoff_base, the next twice that, and so on, capped at retry_backoff_max. Once attempts reaches max_attempts the task is abandoned (failed) rather than retried. A missing handler is non-retriable and fails on the first attempt regardless of max_attempts.

The backoff knobs are worker-level defaults on WorkerOptions:

use std::time::Duration;
use umbral_tasks::{run_worker, WorkerOptions};
 
run_worker(WorkerOptions {
    retry_backoff_base: Duration::from_secs(2),     // default
    retry_backoff_max:  Duration::from_secs(5 * 60), // default
    ..WorkerOptions::default()
}).await;

The orphan-reclaim path applies the same backoff: a reclaimed-but-not-exhausted task is pushed forward by run_at too, so a job that keeps crashing its worker doesn't get retried in a tight loop.

Per-task timeout

The worker wraps each handler invocation in tokio::time::timeout. A handler that overruns WorkerOptions::task_timeout is cancelled and recorded as a retriable failure (backed off via run_at, or abandoned if max_attempts is exhausted) - it never holds its claim until the visibility timeout fires.

use std::time::Duration;
use umbral_tasks::{run_worker, WorkerOptions};
 
run_worker(WorkerOptions {
    task_timeout: Some(Duration::from_secs(30)), // default: 5 minutes; None disables
    ..WorkerOptions::default()
}).await;

This is distinct from the visibility timeout below: task_timeout fails a running handler promptly, whereas the visibility timeout only reclaims a row after a crashed worker stops renewing its lease.

Visibility timeout and at-least-once delivery

When a worker claims a task it stamps started_at = now(). If the worker process crashes mid-handler that row stays in running forever - a silent data loss. The visibility timeout closes this gap.

On every poll the worker first calls reclaim_orphaned_tasks(visibility_timeout): any row whose status = 'running' and started_at < now - visibility_timeout is considered abandoned. The reclaim pass:

• resets the row to pending (clearing started_at) so the next worker poll picks it up again, as long as attempts < max_attempts.
• marks the row failed if attempts >= max_attempts - no infinite retry loop.

The default timeout is 5 minutes (DEFAULT_VISIBILITY_TIMEOUT). Override it when building the worker:

use std::time::Duration;
use umbral_tasks::{run_worker, WorkerOptions};
 
run_worker(WorkerOptions {
    visibility_timeout: Duration::from_secs(30 * 60), // 30 minutes
    ..WorkerOptions::default()
}).await;

You can also call reclaim_orphaned_tasks directly - for example, in a scheduled job or a one-shot management command - with any timeout you choose:

use std::time::Duration;
use umbral_tasks::reclaim_orphaned_tasks;
 
let reclaimed = reclaim_orphaned_tasks(Duration::from_secs(60)).await?;
println!("{reclaimed} orphaned task(s) reclaimed");

Periodic tasks (beat)

For recurring work - a nightly cleanup, an hourly digest, a job every five minutes - register a periodic schedule and run the beat process. Beat is the periodic-task scheduler; it does not run handlers itself. Each time a schedule is due, beat enqueues a normal task row; the worker drains it. Run them as two separate processes: the worker that executes handlers, and beat that schedules them.

Register schedules on the plugin builder with .periodic(name, schedule, task, payload):

use std::time::Duration;
use umbral_tasks::{Schedule, TasksPlugin};
 
App::builder()
    .plugin(
        TasksPlugin::default()
            // Cron: midnight every day.
            .periodic(
                "cleanup",
                Schedule::cron("0 0 * * *"),
                "cleanup_task",
                serde_json::json!({ "older_than_days": 30 }),
            )
            // Fixed interval: every hour.
            .periodic(
                "heartbeat",
                Schedule::every(Duration::from_secs(3600)),
                "heartbeat_task",
                serde_json::json!({}),
            ),
    )
    .build()?;

name is the schedule's stable key - one periodic_task row per name. Re-registering with the same name updates the existing row (its task, payload, and schedule) instead of duplicating it; next_run is only recomputed when the schedule string actually changes, so an unchanged redeploy never starves the task.

Schedule format

The schedule persists as a single string column ("cron:0 0 * * *" / "every:3600"). A schedule that yields no further fire time disables the row (enabled = false) rather than spinning.

Running beat

# Run the scheduler continuously alongside the worker.
cargo run -- tasks-beat
 
# Sync schedules, run one tick, and exit (tests / cron-driven beats).
cargo run -- tasks-beat --once

On startup beat syncs the registered schedules to periodic_task rows, then each tick (default every 5s) enqueues every due schedule.

Task results & status

A handler can return any serializable value, not just Ok(()). On success the worker serializes the return value to JSON and stores it in the task row's result column - the result backend payload. Query a task's outcome by the id enqueue returned:

use umbral_tasks::{enqueue, register_handler, task_status, await_result, TaskState, EnqueueOptions};
use std::time::Duration;
 
// A handler that returns a value (any `R: Serialize`):
register_handler("sum", |payload: &str| async move {
    let nums: Vec<i64> = serde_json::from_str(payload).map_err(|e| e.to_string())?;
    Ok::<i64, String>(nums.iter().sum())
});
 
let id = enqueue("sum", vec![1, 2, 3], EnqueueOptions::default()).await?;
 
// Non-blocking: snapshot the current state (None if the row was drained).
let status = task_status(id).await?.expect("task exists");
assert_eq!(status.state, TaskState::Pending); // not run yet
 
// Blocking with a timeout: poll until terminal.
let status = await_result(id, Duration::from_secs(5)).await?;
assert_eq!(status.state, TaskState::Success);
assert_eq!(status.result, Some(serde_json::json!(6)));

task_status returns a TaskStatus { id, name, state, result, error, attempts, max_attempts, run_at, started_at, completed_at, created_at }. state is a TaskState - Pending, Running, Success, Failed, or Retrying (a row that failed an attempt but still has retries left). await_result polls every 50ms until the state is terminal (Success/Failed) or timeout elapses, in which case it returns TaskError::Timeout carrying the last observed status.

See arch.md and plugins/umbral-tasks/src/lib.rs for the design rationale.

Running it in production

The queue is only as real as the process that drains it. enqueue writes a row and returns — if nothing runs tasks-worker, your jobs sit in the table forever, which looks exactly like "the feature silently doesn't work". So the worker is a service, alongside your web process:

A compose deployment, extending the one in Going to production (db → migrate → web) with the two task services:

services:
  migrate:
    image: myapp:latest
    command: ["myapp", "migrate"]
    depends_on: { db: { condition: service_healthy } }
    restart: "no"                       # one-shot; must finish before web/worker start
 
  web:
    image: myapp:latest
    command: ["myapp", "serve"]
    depends_on: { migrate: { condition: service_completed_successfully } }
 
  worker:                               # drains the queue
    image: myapp:latest                 # SAME image — it needs your #[task] fns
    command: ["myapp", "tasks-worker"]
    depends_on: { migrate: { condition: service_completed_successfully } }
    restart: unless-stopped
    deploy: { replicas: 2 }             # scale freely
 
  beat:                                 # fires periodic tasks — exactly one
    image: myapp:latest
    command: ["myapp", "tasks-beat"]
    depends_on: { migrate: { condition: service_completed_successfully } }
    restart: unless-stopped

Three things that bite people:

• Same image, same binary. The worker executes your #[task] functions, so it must be the app binary, not a generic runner. A worker built from a stale image runs stale task code — or fails with HandlerNotFound for a task the web process happily enqueues.
• Wait for migrate. Both services touch task_row; starting them before the one-shot migrate finishes means querying a table that doesn't exist yet.
• A deploy changes the worker, too. Roll web and worker together. A task enqueued by new web code and picked up by an old worker is the HandlerNotFound case.

--once (tasks-worker --once, tasks-beat --once) drains a single batch and exits — the shape for a cron-driven or CI run when you don't want a resident process.

Admin visibility

Operators browse and inspect the task queue in the admin. umbral_tasks::admin_model() returns a read-only AdminModel for the task_row table - register it on the admin alongside your own models:

use umbral_admin::AdminPlugin;
 
App::builder()
    .plugin(TasksPlugin::default())
    .plugin(AdminPlugin::default().register(umbral_tasks::admin_model()))
    .build()?;

Register periodic_admin_model() alongside it to see the schedules — the queue tells you what ran, not what is scheduled:

AdminPlugin::default()
    .register(umbral_tasks::admin_model())           // the queue
    .register(umbral_tasks::periodic_admin_model())  // the schedules

It lists each periodic task's cron expression, next_run, last_run, and whether it's enabled — with an Enable / disable bulk action, so pausing a runaway schedule at 3am doesn't need a psql session. enabled is the only editable column: everything else is declared in code and re-synced at boot, so hand-editing it here would be silently reverted on the next deploy.

The queue list shows the columns an operator wants at a glance - id, name, status, priority, attempts / max_attempts, run_at, completed_at, created_at - with status / priority filters and a name / status search box. Every column is read-only: rows are written by enqueue and the worker, never hand-authored.

Retrying a failed task

The admin model attaches a "Retry selected" bulk action. Select one or more failed rows and run it to re-queue them: each failed task is reset to pending with run_at = now (immediately eligible), its error cleared, and its attempts counter reset to 0 so it gets a fresh max_attempts budget. The action only touches rows currently in the failed state - a pending, running, or succeeded task is left untouched, so you can't disturb a task a worker is mid-flight on.

The same logic is available programmatically:

// Re-queue task #42 if (and only if) it is currently failed.
let requeued: bool = umbral_tasks::retry_task(42).await?;

The admin_model() function and the umbral-admin dependency are gated behind the admin feature (on by default). A tasks-only app that doesn't run the admin builds with default-features = false and never pulls the admin into its dependency graph:

umbral-tasks = { version = "0.0.1", default-features = false }

Combining with signals

// in your signup signal handler
enqueue("email.send_welcome", WelcomeEmailPayload { user_id }, EnqueueOptions::default()).await?;

What is NOT in v1

• FOR UPDATE SKIP LOCKED. The worker uses a conditional UPDATE claim that is correct on both SQLite and Postgres (a concurrent worker whose UPDATE matches zero rows loses the race cleanly). SELECT … FOR UPDATE SKIP LOCKED is a future optimization that avoids the wasted candidate SELECT, not a correctness fix.
• Per-task backoff / timeout persistence. EnqueueOptions accepts the fields but the worker applies worker-level defaults in v1 (see the callout above). #82.
• Status enum. Status is stored as a plain string (pending, running, succeeded, failed) because the M3 derive doesn't support enum SqlType yet.

The spec outline lives in docs/specs/outlines/. Source: plugins/umbral-tasks/src/lib.rs.