Jobs (Active Job + Solid Queue, done the Oban way → kernel/jobs/) #

TL;DR

• A persistent, database-backed queue in src/kernel/jobs/, modeled on Oban: states, retries with backoff, uniqueness, snooze/cancel/discard, cron, Stager/Lifeline/Pruner plugins, safe on every node.
• Args hold ids, never records or secrets (args are visible in /admin/jobs); enqueue inside the transaction that creates the data.
• Jobs are idempotent (at-least-once delivery); size MAX_ATTEMPTS to the work, and transactional mail expires instead of retrying for weeks.
• Test with drain_jobs, assert_enqueued, perform_job; no sleeps.

Read the rest when adding a worker, a recurring job, a queue, or touching the kernel queue.

Use the part. The jobs parts are this guide built, tested on SQLite and Postgres and security-reviewed, with more on top: batches, rate-limited queues, priorities, timeouts, and args and errors redacted wherever they're shown or searched. When the product needs this, add the part rather than writing it from this guide: cargo offrails add-part jobs (dependencies come first; workflow). Keep its // SECURITY: markers and the tests its CHECKS.md cites. This guide stays the spec: hand-write from it only when no part fits, or the human declined the part.

Background work runs from a persistent, database-backed queue that you generate into src/kernel/jobs/. It's modeled on Elixir's Oban:

• jobs are rows with an explicit state machine;
• finished jobs are kept (and pruned later), so you have history to inspect;
• queues have concurrency limits and can be paused;
• a dashboard and a CLI sit on top (jobs-dashboard).

The same queue runs on SQLite and Postgres, so you need no Redis and no extra service. Everything below was compiled and tested on both (sea-orm 2.0.4, axum 0.8.9, croner 4.0.1, rand 0.10.3; 2026-10-12).

Time types: scheduling and backoff use chrono::Duration (TimeDelta; schedule_in, Unique::period, WorkerConfig::backoff), because they're added to DateTimeUtc values stored in the database. Timeouts and shutdown grace use std::time::Duration (Worker::TIMEOUT, WorkerConfig::timeout, shutdown_grace), because tokio consumes them.

Files #

src/kernel/jobs/mod.rs       entities (jobs, job_queues, job_nodes, recurring_executions), JobState, Outcome,
                             JobInfo, WorkerConfig, DispatchFn, the builder (JobDefaults, JobBuilder, Unique), insert_all
src/kernel/jobs/executor.rs  WorkerOptions, run (producers + shutdown), claim, record, drain (tests)
src/kernel/jobs/plugins.rs   Recurring, Heartbeat, stage, rescue_orphans, prune, cron_tick, next_run/previous_run
src/kernel/jobs/admin.rs     queries + actions for the dashboard and the CLI (jobs-dashboard.md)
src/kernel/jobs/web.rs       /admin/jobs (jobs-dashboard.md)
src/kernel/ui.rs             shared chrome for kernel pages (jobs dashboard, /dev, mailbox)
src/jobs/mod.rs              the app's Worker trait, enqueue, dispatch, worker_config, dispatch_fn
src/jobs/recurring.rs        SCHEDULE
src/jobs/<name>.rs           one file per worker (deliver_email.rs, cleanup.rs from `new`)

The kernel never names an app type: it gets the app's dispatcher as a DispatchFn and per-worker settings through a WorkerConfigFn.

The state machine #

                insert (scheduled_at > now)        insert (due now)
                         │                               │
                         ▼                               ▼
                    scheduled ──── Stager (due) ───▶ available ◀──── retry / run now (dashboard)
                                                         │
                   retryable ──── Stager (due) ──────────┤ claim (attempt += 1)
                       ▲                                 ▼
                       │  Err, attempt < max          executing ──── Ok(Done) ──────────▶ completed
                       └──────────────────────────────── │ ──────── Ok(Snooze) ─────────▶ scheduled
                                                         │ ──────── Ok(Cancel) / cancel ─▶ cancelled
                                                         └ Err at max / Ok(Discard) ────▶ discarded

completed, cancelled and discarded are final. Rows stay until the Pruner deletes them (APP_JOBS_PRUNE_DAYS, default 7). The dashboard needs that history, and so do you when debugging.

A job becomes runnable only when it is available and scheduled_at <= now. The Stager flips scheduled/retryable rows to available once their scheduled_at passes; operator actions ("retry", "run now") set both.

Tables (migration create_jobs, generated by generators/new.md; every app has jobs) #

jobs:
  id            big_pk_auto
  state         string            // CHECK in (available, scheduled, executing, retryable, completed, discarded, cancelled)
  queue         string            default "default"
  worker        string            // Worker::NAME. Permanent: rows reference it
  args          json              // serde_json::Value of the worker struct
  meta          json              default {}   // cron key, snoozed count, unique_states, anything the app adds
  tags          json              default []   // ["digest", "user:42"]: filterable in the dashboard
  errors        json              default []   // [{attempt, at, error}], one per failed attempt
  attempt       integer           default 0
  max_attempts  integer           default 20
  priority      small_integer     default 0    // 0 = highest … 9 = lowest
  unique_key    string_null       // see "Uniqueness"
  attempted_by  string_null       // node id: "<hostname>:<pid>"
  inserted_at   timestamp_with_time_zone
  scheduled_at  timestamp_with_time_zone     // when it becomes runnable
  attempted_at  timestamp_with_time_zone_null
  completed_at  timestamp_with_time_zone_null
  cancelled_at  timestamp_with_time_zone_null
  discarded_at  timestamp_with_time_zone_null
index jobs_fetch_idx     (state, queue, priority, scheduled_at, id)
index jobs_worker_idx    (worker, state)
index jobs_inserted_idx  (inserted_at)
unique index jobs_unique_key_key (unique_key)   // NULLs never collide (verified on SQLite and Postgres)

job_queues:     name string pk, paused boolean default false, updated_at timestamptz      // pause/resume across nodes
job_nodes:      node string pk, queues json ({"default": 10}), started_at, heartbeat_at   // Heartbeat + Lifeline + dashboard
recurring_executions: id big_pk_auto, key string, run_at timestamptz; unique index (key, run_at)

Store state as a DeriveActiveEnum string (rename_all = "snake_case") with the CHECK constraint, like any enum (models). The non-auto-increment string primary keys need #[sea_orm(primary_key, auto_increment = false)].

The Worker trait and dispatcher (src/jobs/mod.rs) #

The trait and dispatcher live in the app because they know AppState.

pub use crate::kernel::jobs::{Inserted, JobBuilder, JobInfo, Outcome, Unique, UniqueStates};

pub trait Worker: Serialize + DeserializeOwned + Send + 'static {
    const NAME: &'static str;                          // permanent, snake_case: "send_welcome_email"
    const QUEUE: &'static str = "default";
    const MAX_ATTEMPTS: i32 = 20;
    const PRIORITY: i16 = 0;
    const TIMEOUT: Duration = Duration::from_secs(300); // std::time::Duration

    /// Default uniqueness for every insert of this worker.
    fn unique() -> Option<Unique> { None }

    fn backoff(attempt: i32) -> chrono::Duration { kjobs::default_backoff(attempt) }

    fn perform(self, job: &JobInfo, state: &AppState)
        -> impl Future<Output = anyhow::Result<Outcome>> + Send;

    /// The builder with this worker's defaults.
    fn into_job(self) -> JobBuilder where Self: Sized {
        JobBuilder::new(
            JobDefaults { worker: Self::NAME, queue: Self::QUEUE, max_attempts: Self::MAX_ATTEMPTS,
                          priority: Self::PRIORITY, unique: Self::unique() },
            serde_json::to_value(&self),
        )
    }
}

/// `perform_later` with the worker's defaults. Pass the transaction that created the job's data.
pub async fn enqueue<W: Worker>(db: &impl ConnectionTrait, w: W) -> Result<Inserted, AppError> {
    Ok(w.into_job().insert(db).await?)
}

/// Explicit, compile-checked registry: every worker appears here exactly once.
pub async fn dispatch(state: AppState, worker: String, args: Value, job: JobInfo) -> anyhow::Result<Outcome> {
    match worker.as_str() {
        DeliverEmailJob::NAME => run::<DeliverEmailJob>(&state, args, &job).await,
        CleanupJob::NAME      => run::<CleanupJob>(&state, args, &job).await,
        other => Ok(Outcome::Discard(format!("unknown worker: {other}"))),
    }
}

/// Same match as `dispatch`, for the kernel's timeout and backoff.
pub fn worker_config(worker: &str) -> WorkerConfig {
    match worker {
        DeliverEmailJob::NAME => config::<DeliverEmailJob>(),
        CleanupJob::NAME      => config::<CleanupJob>(),
        _ => WorkerConfig::default(),
    }
}

fn config<W: Worker>() -> WorkerConfig { WorkerConfig { timeout: W::TIMEOUT, backoff: W::backoff } }

async fn run<W: Worker>(state: &AppState, args: Value, job: &JobInfo) -> anyhow::Result<Outcome> {
    match serde_json::from_value::<W>(args) {
        Ok(w) => w.perform(job, state).await,
        Err(e) => Ok(Outcome::Discard(format!("args don't deserialize: {e}"))),   // retrying won't fix it
    }
}

/// The dispatcher as the function the kernel calls.
pub fn dispatch_fn(state: AppState) -> DispatchFn {
    Arc::new(move |worker, args, job| Box::pin(dispatch(state.clone(), worker, args, job)))
}

In the kernel (kernel/jobs/mod.rs):

pub enum Outcome {
    Done,
    Snooze(chrono::Duration),  // not now: back to `scheduled`; doesn't consume an attempt
    Cancel(String),            // the work no longer matters (record deleted) → cancelled
    Discard(String),           // permanently failed, retrying can't help (invalid address) → discarded
}
// Err(e) → retryable with backoff, or discarded at max_attempts.

pub struct JobInfo { pub id: i64, pub attempt: i32, pub max_attempts: i32, pub inserted_at: DateTimeUtc, pub tags: Vec<String>, pub meta: Value }
pub struct WorkerConfig { pub timeout: std::time::Duration, pub backoff: fn(i32) -> chrono::Duration }   // Default: 300s, default_backoff
pub type DispatchFn = Arc<dyn Fn(String, Value, JobInfo) -> BoxFuture<anyhow::Result<Outcome>> + Send + Sync>;
pub type WorkerConfigFn = fn(&str) -> WorkerConfig;

NAME strings are permanent. Rows already in the queue (and the dashboard's history) reference them. Renaming a struct is fine, but keep its NAME. When changing a struct's fields, keep old rows decodable: #[serde(default)] on new fields, or #[serde(flatten)] when wrapping the old shape (see DeliverEmailJob in mailers).

Enqueueing: the builder #

// defaults
jobs::enqueue(&txn, NotifyAuthorJob { comment_id }).await?;

// everything else
SendWeeklyDigestJob { user_id }
    .into_job()
    .queue("digests")                                    // override W::QUEUE
    .priority(3)                                         // clamped to 0..=9
    .schedule_in(chrono::Duration::hours(1))             // or .schedule_at(dt)
    .max_attempts(5)                                     // at least 1
    .tags(["digest", &format!("user:{user_id}")])        // replaces; .tag("x") appends
    .meta(json!({"source": "settings_page"}))            // merged into meta
    .unique(Unique::args().period(chrono::Duration::hours(24)))
    .insert(&txn)                                        // &DatabaseConnection or &DatabaseTransaction
    .await?;                                             // -> Inserted { id, duplicate: bool }

• JobBuilder is not generic (it holds the serialized args), so mixed workers batch together and recurring entries can be fn() -> JobBuilder.
• insert takes any ConnectionTrait. Pass the transaction that created the data the job needs, so the job is committed with it (after-commit semantics, for free).
• state is scheduled if scheduled_at > now, else available. Timestamps come from Utc::now() in Rust, never the database clock.
• Args that fail to serialize make insert return Err (no row).
• kernel::jobs::insert_all(db, Vec<JobBuilder>) does a batch insert (fan-out) in one statement. It refuses unique jobs (insert those one by one).
• An insert wakes this process's producers immediately (a tokio::sync::Notify), so inline-worker apps feel instant.

Uniqueness #

Unique::args(), Unique::args_keys(&["user_id"]) or Unique::worker(), plus optional .period(chrono::Duration) and .states(UniqueStates::Incomplete | UniqueStates::All):

1. unique_key = hex(sha256("{worker}|{queue}|{selected args as JSON}|{bucket}")), where bucket = floor(now / period) (empty without a period: unique "while it exists"). serde_json's map is a BTreeMap, so the JSON is canonical.
2. Insert with job::Entity::insert(row).on_conflict_do_nothing_on([job::Column::UniqueKey]).exec(db). In sea-orm 2 that returns TryInsertResult::{Inserted, Conflicted, Empty}, not a RecordNotInserted error (and .do_nothing() is deprecated). On Conflicted, load the row by unique_key and return Inserted { id: existing, duplicate: true }. Atomic on both backends, no advisory locks.
3. The builder records meta.unique_states = "incomplete" | "all". With Incomplete (the default), the kernel sets unique_key = NULL when the job reaches a final state (worker outcome, operator cancel, Lifeline discard), so an identical job can be inserted afterwards. With All, the key lives until the Pruner deletes the row.

Use uniqueness for "at most one pending digest per user", or a debounce ("re-index this post at most once a minute"). It's not a substitute for idempotency.

Kernel spec #

Worker runtime (executor.rs) #

pub struct WorkerOptions {
    pub queues: Vec<(String, usize)>,            // APP_QUEUES / app worker --queues
    pub node: String,                            // kernel::jobs::node_id(): "<hostname>:<pid>"
    pub prune_after: chrono::Duration,           // APP_JOBS_PRUNE_DAYS
    pub shutdown_grace: std::time::Duration,     // APP_JOBS_SHUTDOWN_GRACE
    pub schedule: &'static [Recurring],          // crate::jobs::recurring::SCHEDULE
}
pub async fn run(db: DatabaseConnection, opts: WorkerOptions, dispatch: DispatchFn,
                 config: WorkerConfigFn, shutdown: impl Future<Output = ()>);

run sends one heartbeat (registers the node, reads paused queues), spawns one producer per queue and the plugin loop, then waits for shutdown (Ctrl-C / SIGTERM).

Each producer (concurrency limit, a Semaphore of that size):

1. Paused? Skip fetching while the queue is in the paused set (refreshed by the Heartbeat every second). In-flight jobs keep running.
2. Fetch: free = available permits. Select up to free ids where state = 'available' AND queue = ? AND scheduled_at <= now, ordered by priority, scheduled_at, id.
3. Claim optimistically, one at a time: update_many set state = 'executing', attempt = attempt + 1, attempted_at = now, attempted_by = node, filtered by id = ? AND state = 'available'. The claim succeeded only if rows_affected == 1. Then load the row. Portable; no SKIP LOCKED.
4. Execute on a JoinSet task holding a permit: tokio::time::timeout(TIMEOUT, AssertUnwindSafe(dispatch(...)).catch_unwind()) (futures::FutureExt). Keep the task's AbortHandle in a shared HashMap<i64, AbortHandle> (cancellation).
5. Record, with the update filtered by id = ? AND state = 'executing' AND attempted_by = node, so a job cancelled or rescued meanwhile isn't overwritten:

Each error is pushed as {attempt, at, error}, redacted first (kernel::redact::text: an error often quotes the URL that failed, credentials and tokens included; errors-and-logging), then capped at 4 KB (start kept) and the last 20 entries. The job.exception log line carries the same redacted text. In final states, clear unique_key unless meta.unique_states == "all". 6. Wake-ups: after a fetch that found nothing, wait for the first of: a local insert (Notify), a job finishing, 1s ± 20% jitter, or shutdown. When every permit is taken, wait for a job to finish.

• default_backoff(attempt) = attempt⁴ + 15s + rand::random_range(0..=30 × attempt) seconds (rand 0.10: rand::random_range). ~16s, ~31s, ~96s, ~271s… 20 attempts cover about three weeks; pick MAX_ATTEMPTS per worker (see mailers' DeliverEmailJob).

Observability #

One tracing span per execution: info_span!("job", job_id, worker, queue, attempt), logging job.start (queue_ms = attempted_at − scheduled_at), then job.stop (duration_ms, outcome), or job.exception (warn! when retryable, error! when discarded). Plugins log debug!(plugin, count) when they changed rows and error! when they fail; Lifeline logs warn!(rescued). That's all the telemetry: the dashboard reads the jobs table directly.

Plugins (plugins.rs; every node runs all of them; all idempotent; no leader election) #

Each is a plain async fn(db, now, …) so tests call it directly instead of waiting for a timer. The plugin loop ticks every second.

Shutdown #

On shutdown: stop fetching, wait up to shutdown_grace (default 25s; must be below the platform's kill timeout) for in-flight jobs, abort the rest, stop the plugins, delete this node's job_nodes row. Aborted jobs stay executing on a node that no longer exists; Lifeline (on any node, or on the next boot) puts them back to available within 30s. Delivery is at-least-once, by design.

Postgres-only optimizations (later, behind a backend match inside the kernel) #

FOR UPDATE SKIP LOCKED claims and LISTEN/NOTIFY wake-ups. Only when measured. The portable path stays the default and stays tested.

Recurring jobs (src/jobs/recurring.rs), the Oban Cron plugin #

use crate::kernel::jobs::Recurring;   // pub struct Recurring { pub key: &'static str, pub cron: &'static str, pub job: fn() -> JobBuilder }

pub const SCHEDULE: &[Recurring] = &[
    Recurring { key: "cleanup",       cron: "0 * * * *",   job: || CleanupJob.into_job() },
    Recurring { key: "weekly_digest", cron: "0 8 * * MON", job: || FanOutWeeklyDigestJob.into_job().priority(5) },
];

job is a non-capturing closure, which coerces to fn() -> JobBuilder, so SCHEDULE stays a const. Exactly how a tick works:

1. run_at = Cron::from_str(cron)?.find_previous_occurrence(&now, true) (croner 4; UTC). An invalid expression logs an error and is skipped (a test asserts every SCHEDULE entry parses).
2. Skip unless now - run_at <= 60s. A node booting at 09:40 does not replay the 08:00 run.
3. In one transaction: insert (key, run_at) into recurring_executions with on_conflict_do_nothing_on([key, run_at]). If nothing was inserted, another node owns this run: roll back.
4. Otherwise insert (job)() with .tag("cron") (appended to the entry's own tags) and .meta(json!({"cron": key})), then commit.

Each run happens exactly once across all nodes. The dashboard's cron page shows each entry's next run (plugins::next_run), its last run (latest recurring_executions.run_at) and the job it enqueued (the newest job tagged cron with meta.cron == key).

Built-in CleanupJob (hourly, generated by new) deletes: expired cache_entries, recurring_executions older than 7 days, and in development mailbox_messages beyond the newest 500. Generators that add framework tables add their cleanup here (sessions, password_resets, broadcasts). Job rows are the Pruner's job, not CleanupJob's.

Queues & configuration #

APP_QUEUES=default:10,mailers:5 sets which queues this process runs, and each queue's concurrency per node. app worker --queues default:10 overrides it per process (dedicated worker roles; see deployment). The default is default:10,mailers:5.

A job inserted into a queue that no node runs just waits. The dashboard flags such queues as "no nodes".

Rules #

1. Args hold ids, not records. { "post_id": 42 }, never a serialized Post. Reload in perform. If the record is gone, return Ok(Outcome::Cancel("post 42 deleted".into())). Exception: DeliverEmailJob carries the rendered email (see mailers).
2. Enqueue inside the transaction that creates the data the job needs.
3. Jobs must be idempotent. At-least-once delivery means a job can run twice (crash after the side effect, before it's marked completed). Guard external side effects with a DB marker (sent_at).
4. Use ? freely, and wrap with .context("...") so the dashboard's error history explains itself.
5. Return Discard only when retrying can't help (bad input). Return Cancel when the work no longer matters. Return Snooze for "not yet" (rate limited by a provider, waiting on another record).
6. Size MAX_ATTEMPTS to the work. The default (20, ~3 weeks) suits syncs and imports. Transactional mail must expire instead: DeliverEmailJob uses 10; mailers::deliver_later_until caps at 5 and discards late mail; token emails use a dedicated job with 5 attempts that discards stale requests.
7. No secrets in args. Args are shown in /admin/jobs and app jobs show. Pass ids and let the worker create or load secrets (tokens are minted in the job, and only their digest is stored; see auth).
8. Long jobs: break them into batches that enqueue the next batch, or a fan-out job that insert_alls one job per item. No job should run longer than its TIMEOUT (5 min default).
9. Tag jobs with what an operator will search for (user:42, import:7).

Gotchas agents get wrong #

• Forgetting to add a new worker to dispatch and worker_config. The job is discarded as "unknown worker". The generator recipe makes it a checklist step.
• Using tokio::spawn for "background work" in a handler. That isn't durable. Use a job.
• Doing the claim as SELECT-then-UPDATE without checking rows_affected, which double-runs jobs.
• Recording a result without filtering by state = 'executing' AND attempted_by = node, which resurrects cancelled or rescued jobs.
• Comparing scheduled_at with a DB-side now(). Pass Utc::now() from Rust.
• Expecting DbErr::RecordNotInserted from on_conflict_do_nothing_on. sea-orm 2 returns TryInsertResult::Conflicted (or, with exec_without_returning, Inserted(0) on some backends: treat both as "not inserted").
• job::ActiveModel::default() is ambiguous in sea-orm 2 (ActiveModelTrait::default vs Default::default). Write <job::ActiveModel as Default>::default() or job::ActiveModel { ..Default::default() }.
• select().offset(n) without .limit(..) is a syntax error on SQLite. Find a cutoff row with offset(n).limit(1) instead.
• Text search over args must search values, not JSON fragments: Postgres jsonb reformats ({"n": 1} with a space). Search LOWER(CAST(args AS text)) LIKE '%needle%' (sea-query: Func::lower(Func::cast_as(Expr::col(..), "text"))) with LikeExpr::new(..).escape('\\').
• Deleting completed jobs on success (the Solid Queue habit). Keep them; the Pruner owns deletion.
• Running Stager/Pruner "only on one node" with leader election. Every plugin is idempotent, so let all nodes run them.

Testing (Oban.Testing equivalents) #

Tests run in manual mode: inserts write rows, and nothing executes until the test says so. There's no inline mode, because enqueueing happens inside the caller's transaction. Running the job there would execute it before the commit, which breaks the after-commit guarantee that production relies on.

Tests assert on outcomes (email sent, row changed). Assert on the queue only when scheduling is the spec ("digest is sent at 8am", "retry in an hour").

Kernel tests (tests/app/kernel_jobs_test.rs, generated with the kernel, run on both backends) use their own DispatchFn (workers named ok, fail, snooze, cancel, discard, panic, hang), so they don't depend on app workers. Required: racing claims run each job exactly once (4 concurrent drains over 30 jobs); retry with backoff then discard at max attempts; snooze; cancel/discard/panic/timeout outcomes; unknown worker and bad args; error caps; errors stored redacted (job_errors_are_stored_redacted: a worker failing with https://deploy:not-a-password@api.example.com/v1/charges?token=not-a-token stores neither secret, keeps the host and the cause readable, and an error search for not-a-password finds nothing); uniqueness (args, args_keys, worker, All, period); insert_all; Lifeline rescues only dead nodes' jobs; Pruner keeps recent and non-final jobs; cron fires once across two concurrent ticks and skips stale due times; the runtime: a worker runs a job and deregisters on shutdown, cancelling an executing job aborts it (a drop guard proves the future was dropped), a paused queue isn't fetched until resumed, shutdown leaves unfinished jobs to Lifeline; and the admin API. To simulate time passing, update the job's scheduled_at into the past, then call stage(db, Utc::now()). Don't stage "in the future", because fetch also requires scheduled_at <= now.