This release enhances workflows with sub-workflows and context sharing, overhauls queue partitioning for better performance, improves dynamic plugins, and adds various usability improvements.
See the v1.6 Upgrade Guide for complete upgrade steps and migration caveats.
🗂️ Sub-Workflows
Workflows gain powerful new capabilities for organizing complex job relationships with two major
enhancements: add_workflow/4 and add_many/4.
Use add_workflow/4 to nest entire workflows within others to create hierarchical job
dependencies. This makes complex workflows more maintainable by grouping related jobs together:
extr_flow =
Workflow.new(workflow_name: "extract")
|> Workflow.add(:extract, WorkerA.new(%{source: "database"}))
|> Workflow.add(:transform, WorkerB.new(%{}), deps: :extract)
# Add sub-workflow as a dependency
Workflow.new()
|> Workflow.add(:setup, WorkerC.new(%{mode: "initialize"}))
|> Workflow.add_workflow(:extract, extr_flow, deps: :setup)
|> Workflow.add(:finalize, WorkerC.new(%{}), deps: :extract)
Workflows can depend on other workflows, and downstream deps will wait until the sub-workflow completes before executing.
Need to run similar jobs in parallel? Use add_many/4 to add multiple jobs with a single
dependency name:
# Add multiple email jobs that can run in parallel
email_jobs = Enum.map(users, &EmailWorker.new(%{user_id: &1.id}))
workflow =
Workflow.new()
|> Workflow.add_many(:emails, email_jobs)
|> Workflow.add(:report, ReportWorker.new(), deps: :emails)
The add_many/4 step creates a sub workflow from either a list or a map, and the full recorded
results can be extracted with a single call:
def process(job) do
map_of_results = Workflow.all_recorded(job, with_subs: true)
end
🖼️ Context and Workflow Status
Workflows that rely on common data can now share data without duplicating arguments using
put_context/3:
workflow =
Workflow.new()
|> Workflow.put_context(%{user_id: 123, app_version: "1.2.3"})
|> Workflow.add(:job_a, WorkerA.new(%{}))
|> Workflow.add(:job_b, WorkerB.new(%{}))
# Later in a worker:
def process(job) do
context = Workflow.get_context(job)
# Use context map...
end
It's now easier to check workflow progress with status/1, which provides execution stats:
%{
total: 5,
state: "executing",
counts: %{completed: 3, executing: 2},
duration: 15_620
} = Workflow.status(workflow_id)
🧰 Queue Partitioning Overhaul
Queue partitioning is completely redesigned for dramatic performance improvements. Jobs are now assigned partition keys on insert rather than at runtime, enabling more efficient querying and eliminating head-of-line blocking when one partition has many jobs.
The new design has none of the issues of the previous solution:
- Job processing is completely fair. Jobs from a single partition can't block processing of other partitions after bulk insert. No priority or scheduling workarounds are necessary.
- Querying in partitioned queues relies on a single, partial index
- Partitioning uses a single, optimized query without any unions or dynamic sections. That allows ecto to prepare and cache a single plan for faster fetching and less memory usage.
In a benchmark of 10k jobs spread across 20 partitions (200k jobs), processing took 17s in v1.6, down from 360s in v1.5 (20x faster) with far less load on the database.
🧨 Global Burst Mode
Global partitioning gained an advanced feature called "burst mode" that allows you to maximize throughput by temporarily exceeding per-partition global limits when there are available resources.
Each global partition is typically restricted to the configured allowed value. However, with
burst mode enabled, the system can intelligently allocate more jobs to each active partition,
potentially exceeding the per-partition limit while still respecting the overall queue
concurrency.
This is particularly useful when:
- You have many potential partitions but only a few are active at any given time
- You want to maximize throughput while maintaining some level of fairness between partitions
- You need to ensure your queues aren't sitting idle when capacity is available
Here's an example of a queue that will 5 jobs from a single partition concurrently under load, but can burst up to 100 for a single partition when there is available capacity:
queues: [
exports: [
global_limit: [
allowed: 5,
burst: true,
partition: [args: :tenant_id]
],
local_limit: 100
]
]
🍋 Preserve DynamicQueues Updates
DynamicQueues now preserves queue changes made at runtime across application restarts. This brings two key improvements:
- Runtime changes to queues (via Web or CLI) persist until explicitly changed in configuration
- A new
:automaticsync mode that can manage queue deletions based on configuration
# Automatic mode - Deletes queues missing from configuration
config :my_app, Oban,
plugins: [{DynamicQueues, sync_mode: :automatic, queues: [...]}]
In automatic mode, any queue that exists in the database but isn't defined in the configuration will be automatically deleted during startup. This is useful when you want to ensure your runtime queue configuration exactly matches what's defined in your application config.
Now when you pause a queue through the dashboard or change its limits via API, those changes will persist across application restarts until you explicitly update those options in your configuration.
🎨 Decorated Job Enhancements
Decorated jobs gain a few new capabilities. You can now use current_job/0 to access the
underlying job struct from within decorated functions, making it easier to work with job context
or pass job details to other functions. Additionally, you can mark any decorated job as recorded
at runtime with the recorded option, enabling workflow composition and return value access
without separate modules.
defmodule Business do
use Oban.Pro.Decorator
@job queue: :default, recorded: true
def process_account(account_id) do
job = current_job()
IO.puts("Processing account #{account_id} with job #{job.id}")
{:ok, %{processed: true}}
end
end
Enhancements
-
[Workflow] Add
add_workflow/4for creating nested sub-workflows.Sub-workflows simplify organizing and managing complex job dependencies by grouping related jobs:
Workflow.new() |> Workflow.add_workflow(:sub, MyApp.SubWorkflow.new()) |> Workflow.add(:final, FinalWorker.new(%{}), deps: :sub)Downstream dependencies can reference the sub-workflow as a whole, including any jobs that may be dynamically added during runtime.
-
[Workflow] Add
add_many/4function for creating fan-out sub-workflows.This helper enables adding multiple jobs to a workflow with a single name:
email_jobs = Enum.map(users, &EmailWorker.new(%{user_id: &1.id}) workflow = Workflow.new() |> Workflow.add_many(:emails, email_jobs)) |> Workflow.add(:report, ReportWorker.new(), deps: :emails) -
[Workflow] Add sub-workflow metadata and query support.
Support retrieving sub-workflow jobs with new
with_subsoption. This makes it possible to fetch all sub-workflow jobs, including all recordings, with a single call.Workflow.all_recorded(job, with_subs: true) -
[Workflow] Add
put_context/3for sharing values between workflow jobs.The helper inserts a completed job with a recorded value for all other jobs to fetch. This simplifies context sharing and eliminates the need to add the same values as args to all jobs in a workflow.
workflow = Workflow.put_context(workflow, %{id: 123, name: "Alice"}) def process(job) do %{id: id, name: name} = Workflow.get_recorded(job, :context) end -
[Workflow] Add
status/1helper for getting workflow execution information.The new helper simplifies gathering runtime information about a workflow, including the name, total jbs, overall state of the workflow, elapsed duration, state counts, and timestamps.
%{total: 5, counts: %{completed: 5}} = Workflow.status(workflow.id) -
[Workflow] Add
retry_jobs/2for retrying workflow jobs.The new helper function will retry jobs in the workflow and hold jobs with dependencies accordingly.
Workflow.retry_jobs(job) -
[Workflow] Optimize workflow indexes and eliminate containment queries.
A new partial workflow index checks held jobs, workflow ID, and sub-workflow ID at once, eliminating the need for a GIN index on meta and args.
-
[Smart] Add burst mode for global partitioned queues.
Allows partitioned queues to exceed per-partition limits when capacity is available, while still respecting the overall concurrency limit:
config :my_app, Oban, queues: [ exports: [ global_limit: [ allowed: 5, burst: true, partition: [args: :tenant_id] ], local_limit: 100 ] ] -
[Smart] Allow using
:metain partition configuration.Now queue partitioning can use values from the job's metadata, including composition values such as
workflow_id.config :my_app, Oban, queues: [media: [global_limit: [allowed: 1, partition: [meta: :workflow_id]]]] -
[Smart] Overhaul queue partitioning for significant performance improvements.
Adds a generated
partition_keycolumn with a partial index tooban_jobs. Jobs are pre-partitioned on insert based on queue configuration, which simplifies partition queries and provides ~20x faster performance. -
[Smart] Centralize producer refresh and cleanup.
Producer refreshing and cleanup is now centralized to reduce database load. Rather than one transaction per queue per node every 30 seconds, there is now one transaction per node every 30 seconds. This offers significant query savings in systems running numerous queues.
-
[Smart] Provide alternate map hashing to avoid collision.
The use of
phash2/1could cause collisions between values such as uuids, which led to false positives for unique checks. This adds an alternate mechanism for hashing that avoids any such collisions, but it is opt-in for backward compatibility.To switch, set the following compile time config:
config :oban_pro, Oban.Pro.Utils, safe_hash: true -
[Decorator] Add
current_job/0to decorated modules.The
current_job/0function allows decorated jobs to access the underlying job for inspection or to pass as context to other functions such asBackoff, orWorkflowhelpers. -
[Decorator] Support recording as a runtime option.
Any job may now be marked as recorded at runtime, including decorated jobs:
@job queue: :processing, recorded: true def process_account(account_id) do # ... end -
[DynamicPrioritizer] Expand options for finer control.
New
:limitand:max_priorityoptions add control for many jobs are prioritized and their maximum priority level. -
[DynamicQueues] Preserve runtime updates without configuration changes.
Add automatic
sync_modeto insert/update/delete queues based on configuration and prevent overwriting runtime updates to queues unless the configuration changes. -
[DynamicCron] Use optimized cron expression calculation.
The new
last_at/2andnext_at/2cron expression functions are vastly faster and more efficient than the previous implementation. This improves cron job insertion performance inguaranteedmode.
Bug Fixes
-
[Migration] Drop tables entirely during initial down migration
The presence of a
uniq_keycolumn prevents dropping the partitioned table after a rollback. The specific sequence of migrating and backfilling is unlikely to be used in the real world.