Pro Features
Pro layers orchestration, distributed concurrency, runtime operations, and worker ergonomics on top of Oban's durable open-source foundation. Explore every feature, see what's open source versus Pro, and where it lands across Elixir and Python.
Every Pro feature
Oban itself is free and open source. Pick a feature to learn about, and whether it's OSS or Pro on your runtime.
Engine & Concurrency
Composition
Workers & Jobs
Scheduling & Maintenance
Local concurrency limits
Cap how many jobs run per node for each queue. Open-source Oban's core concurrency control, applied independently on every node.
Global concurrency limits
Cap how many jobs run for a queue across the entire cluster, not just per node. Set one global limit and the Pro engine coordinates execution between every connected node.
queues: [
media: [global_limit: 10]
]
Rate limiting
Hold a queue to a fixed budget over a sliding window, such as 500 jobs per minute. Choose fixed, leaky-bucket, or token-bucket algorithms, then partition the limit per tenant or account.
queues: [
api: [
rate_limit: [allowed: 500, period: {1, :minute}]
]
]
Queue partitioning
Split a queue's concurrency or rate limits into independent sub-queues. Each partition gets its own budget, so uneven traffic keeps moving fairly.
queues: [
api: [
global_limit: [allowed: 10, partition: :worker]
]
]
Bulk operations
Insert and ack jobs in batches automatically. The engine groups writes to reduce database operations and keep busy queues moving.
1..10_000
|> Stream.map(&MyWorker.new(%{id: &1}))
|> Oban.insert_all()
Workflows
Compose jobs with arbitrary dependencies for sequential, fan-out, and fan-in execution. Pipelines survive restarts and deploys through database-backed tracking and shared context, and saga-style compensations automatically reverse completed jobs when a workflow is cancelled.
Workflow.new()
|> Workflow.add(:fetch, Fetch.new(%{}))
|> Workflow.add(:render, Render.new(%{}), deps: :fetch)
Cascades
Define workflow steps as cascading functions that flow results into downstream jobs, wiring a pipeline together without manually threading results between jobs.
Workflow.new()
|> Workflow.put_context(%{account_id: account_id})
|> Workflow.add_cascade(:export, {sources, &export/2})
|> Workflow.add_cascade(:archive, &archive/1)
|> Workflow.add_cascade(:notify, ¬ify/1)
Sub-workflows
Compose complex workflows from simpler sub-workflows, making it easier to organize and reuse workflow patterns.
Workflow.new()
|> Workflow.add(:setup, new_setup(%{mode: "initialize"}))
|> Workflow.add_workflow(:extract, extract_flow, deps: :setup)
|> Workflow.add_workflow(:notify, notify_flow, deps: :extract)
|> Workflow.add(:finalize, new_cleanup(%{mode: "cleanup"}), deps: :notify)
Batches
Group related jobs and track progress across all nodes, then fire callbacks on transitions such as a cancellation or the entire batch finishing.
defmodule MyApp.Batch do
use Oban.Pro.Worker
@behaviour Oban.Pro.Batch
@impl Oban.Pro.Batch
def batch_completed(_job) do
Logger.info("BATCH COMPLETE")
:ok
end
end
Chains
Force a sequence of jobs to run one after another, ordered and never overlapping regardless of scheduling, retries, or snoozes.
defmodule MyApp.Chain do
use Oban.Pro.Worker, chain: [by: [args: :account_id]]
@impl Oban.Pro.Worker
def process(job) do
# runs strictly after the previous chained job
end
end
Chunks
Execute multiple jobs within a single function, triggered by total availability or a timeout. Multiple chunks can run in parallel within a single queue.
defmodule MyApp.Chunk do
use Oban.Pro.Chunk,
by: :worker,
size: 100,
timeout: 1_000
@impl true
def process([_ | _] = jobs) do
# handle up to 100 jobs at once
end
end
Backfills
Work through datasets of any size with a chain of small, cursor-paginated jobs. Reprocess records or fill in missing data with optional throttling, avoiding the long transactions and locks of a big migration.
defmodule MyApp.UserBackfill do
use Oban.Pro.Backfill, queue: :backfills, limit: 1_000
@impl Oban.Pro.Backfill
def backfill(cursor, _extra) do
MyApp.User
|> where([user], is_nil(user.migrated_at))
|> Cursor.update_all(cursor, set: [migrated_at: ^DateTime.utc_now()])
end
end
Reliable jobs, queues, and retries
The open-source foundation of durable jobs, isolated queues, automatic retries with backoff, and at-least-once execution.
Scheduled and delayed jobs
Enqueue jobs to run at a specific time or after a relative delay, down to the second.
Job priorities
Assign each job a priority so higher-priority work runs ahead of the rest within its queue.
Job tags
Label jobs with tags for filtering, searching, and grouping in queries, metrics, and the dashboard.
Cancelling and retrying jobs
Cancel running or scheduled jobs and retry discarded ones on demand, individually or in bulk.
Snoozing
Postpone an executing job by returning a snooze, deferring it without burning an attempt.
Telemetry instrumentation
Built-in telemetry events for every job and plugin, ready to wire into metrics, logging, and tracing.
Cron scheduling
Schedule recurring jobs with standard cron expressions and configurable options, centrally, across all nodes.
Unique jobs
Prevent duplicate jobs with configurable uniqueness over a time period. It is open source on Elixir and included in Pro on Python.
Decorators
Turn a regular function into a background job with a decorator, no separate worker module required. Decorators are Pro on Elixir and open source on Python.
use Oban.Pro.Decorator
@job queue: :mailers, max_attempts: 3
def deliver_welcome(user_id) do
# runs in the background when called
end
Structured args
Define typed args with compile-time structs and validation, so malformed payloads fail fast before a job ever runs.
args_schema do
field :id, :id, required: true
field :name, :string, required: true
field :mode, :enum, values: ~w(on off paused)a
field :safe, :boolean, default: false
embeds_one :address, required: true do
field :street, :string
field :number, :integer
field :city, :string
end
end
Recorded output
Capture a job's return value and persist it alongside the job. Read results back later or hand them off to downstream jobs.
defmodule MyApp.Worker do
use Oban.Pro.Worker, recorded: true
@impl Oban.Pro.Worker
def process(%Job{}) do
{:ok, %{total: 42}}
end
end
External recorded storage
Offload recorded output to external storage such as S3, Tigris, or Redis instead of the jobs table. Stored output is fetched transparently and deleted automatically when jobs are pruned.
use Oban.Pro.Worker,
recorded: [
storage: {MyApp.S3, bucket: "job-output"},
limit: {32, :mb}
]
Execution hooks
Run code before and after every job in a worker, with access to the result. Global hooks keep lifecycle behavior in one place rather than scattered between perform functions.
defmodule MyApp.Worker do
use Oban.Pro.Worker
@impl Oban.Pro.Worker
def after_process(_state, %Job{} = job) do
# runs after every job in the worker
end
end
Encrypted args
Transparently encrypt sensitive args at rest, keeping secrets or PII out of plaintext in the database.
use Oban.Pro.Worker,
encrypted: [key: {MyApp.Vault, :encryption_key, []}]
Execution deadlines
Set a wall-clock deadline for a job. When the deadline passes, the job is automatically cancelled rather than running forever.
use Oban.Pro.Worker, deadline: {30, :seconds}
Inline cron
Declare a cron schedule directly on a worker or decorated function, keeping the schedule next to the code it runs. Annotations are discovered automatically at boot.
defmodule MyApp.NightlyCleanup do
use Oban.Pro.Worker, cron: "0 3 * * *"
@impl Oban.Pro.Worker
def process(_job) do
# runs every night at 3am
end
end
Awaiting signals
Pause jobs mid-execution to wait for an external decision and resume when a signal arrives. This turns workers into durable state machines that wait for human approval, callbacks, or out-of-band events without holding a worker slot open.
@impl Oban.Pro.Worker
def process(job) do
case Worker.await_signal(wait_for: {1, :hour}) do
{:ok, %{decision: "approved"}} -> charge_card()
{:ok, %{decision: "rejected"}} -> {:cancel, :rejected}
{:error, :timeout} -> {:cancel, :no_decision}
end
end
Relay
Insert a job and await its result synchronously from any node. Relay gives you persistent distributed tasks with results forwarded back across nodes.
%{id: 1}
|> MyApp.Worker.new()
|> Oban.Pro.Relay.async()
|> Oban.Pro.Relay.await()
Queue control
Pause, resume, and scale queues at runtime locally or across all nodes. Changes apply immediately but reset on restart, unlike Pro's persisted queues.
Pruning
Automatically delete old completed, cancelled, and discarded jobs to keep the jobs table small.
Rescuing
Return jobs orphaned by a crashed or restarted node back to available so they run again.
Reindexer
Periodically rebuild jobs-table indexes to fight bloat on busy Postgres databases.
Pro Cron
Insert, update, delete, and pause scheduled jobs at runtime with cluster-wide coordination, timezones, and scheduling guarantees.
cron: {
Oban.Pro.Cron,
sync_mode: :automatic,
crontab: [
{"0 * * * *", MyApp.HourlyWorker}
]
}
Pro Queues
Start, stop, pause, and reconfigure queues at runtime, with changes persisted across restarts. Optionally pin queues to specific nodes.
queues: {
Oban.Pro.Queues,
queues: [default: 10, media: [limit: 20]]
}
Pro Pruner
Tune job retention with persisted, runtime-editable rules. Rules match any combination of queue, worker, and state, retain jobs by age or count, and apply in a deterministic order.
pruner: {
Oban.Pro.Pruner,
rules: [
[name: "events", queue: :events, max_age: {10, :minutes}],
[name: "discarded", state: :discarded, max_age: {7, :days}]
]
}
Job archiving
Keep pruned jobs in an archive table instead of deleting them, copied in a single transaction so nothing is lost between tables. Query archived rows later as regular Oban jobs.
rules: [
[name: "audit",
worker: MyApp.AuditWorker,
max_age: {1, :week},
archive: true]
]
Dynamic Prioritizer
Automatically raise the priority of older jobs over time so low-priority work is eventually processed.
plugins: [
{Oban.Pro.Plugins.DynamicPrioritizer, interval: :timer.minutes(1)}
]
Pro Lifeline
Rescue jobs orphaned by a crashed node without duplicate execution, retry exhausted attempts, and unstick workflows with missing dependencies.
lifeline: Oban.Pro.Lifeline
Dynamic Scaler
Horizontally scale worker nodes based on predictive throughput. Add capacity during activity spikes and scale back during lulls.
plugins: [
{Oban.Pro.Plugins.DynamicScaler,
scaler: {MyApp.Cloud, min: 1, max: 5}}
]
Engine & Concurrency
Composition
Workers & Jobs
Scheduling & Maintenance
Local concurrency limits
Cap how many jobs run per node for each queue. Open-source Oban's core concurrency control, applied independently on every node.
Global concurrency limits
Cap how many jobs run for a queue across the entire cluster, not just per node. Set one global limit and the Pro engine coordinates execution between every connected node.
Read the docs
Rate limiting
Hold a queue to a fixed budget over a sliding window, such as 500 jobs per minute. Choose fixed, leaky-bucket, or token-bucket algorithms, then partition the limit per tenant or account.
Read the docs
Queue partitioning
Split a queue's concurrency or rate limits into independent sub-queues. Each partition gets its own budget, so uneven traffic keeps moving fairly.
Read the docs
Bulk operations
Insert and ack jobs in batches automatically. The engine groups writes to reduce database operations and keep busy queues moving.
Read the docs
Multi-process execution
Run jobs across multiple OS processes to scale CPU bound work across all cores instead of contending on a single thread.
Read the docs
Workflows
Compose jobs with arbitrary dependencies for sequential, fan-out, and fan-in execution. Pipelines survive restarts and deploys through database-backed tracking and shared context, and saga-style compensations automatically reverse completed jobs when a workflow is cancelled.
Read the docs
Cascades
Define workflow steps as cascading functions that flow results into downstream jobs, wiring a pipeline together without manually threading results between jobs.
Read the docs
Sub-workflows
Compose complex workflows from simpler sub-workflows, making it easier to organize and reuse workflow patterns.
Read the docs
Reliable jobs, queues, and retries
The open-source foundation of durable jobs, isolated queues, automatic retries with backoff, and at-least-once execution.
Scheduled and delayed jobs
Enqueue jobs to run at a specific time or after a relative delay, down to the second.
Job priorities
Assign each job a priority so higher-priority work runs ahead of the rest within its queue.
Job tags
Label jobs with tags for filtering, searching, and grouping in queries, metrics, and the dashboard.
Cancelling and retrying jobs
Cancel running or scheduled jobs and retry discarded ones on demand, individually or in bulk.
Snoozing
Postpone an executing job by returning a snooze, deferring it without burning an attempt.
Telemetry instrumentation
Built-in telemetry events for every job and plugin, ready to wire into metrics, logging, and tracing.
Cron scheduling
Schedule recurring jobs with standard cron expressions and configurable options, centrally, across all nodes.
Unique jobs
Prevent duplicate jobs with configurable uniqueness over a time period. It is open source on Elixir and included in Pro on Python.
Decorators
Turn a regular function into a background job with a decorator, no separate worker module required. Decorators are Pro on Elixir and open source on Python.
Read the docs
Awaiting signals
Pause jobs mid-execution to wait for an external decision and resume when a signal arrives. This turns workers into durable state machines that wait for human approval, callbacks, or out-of-band events without holding a worker slot open.
Relay
Insert a job and await its result synchronously from any node. Relay gives you persistent distributed tasks with results forwarded back across nodes.
Read the docs
Queue control
Pause, resume, and scale queues at runtime locally or across all nodes. Changes apply immediately but reset on restart, unlike Pro's persisted queues.
Pruning
Automatically delete old completed, cancelled, and discarded jobs to keep the jobs table small.
