This release introduces saga-style workflow compensations, cursor-based backfills, inline cron scheduling, external recorded storage, persistent pruning rules with automatic archiving, and unified configuration for queues and services.
See the v1.8 Upgrade Guide for complete upgrade steps and migration caveats.
↩️ Workflow Compensations
Workflows now offer compensations, where completed jobs are automatically reversed after a workflow is cancelled or discarded. Compensations run as dedicated jobs in reverse dependency order, including for sub-workflows, grafts, and other dynamically added jobs.
Compensations can be declared explicitly, as a function capture passed as a compensate option
for any workflow shape, including cascades:
Workflow.new()
|> Workflow.add_cascade(:hold, &Inventory.reserve/1, compensate: &Inventory.release/1)
|> Workflow.add_cascade(:charge, &Billing.charge/1, deps: :hold, compensate: &Billing.refund/1)
Alternatively, worker-module style workflows that declare a compensate/1 callback will be
compensated automatically:
@impl Oban.Pro.Workflow
def compensate(%Job{args: args}) do
Billing.refund(args["charge_id"])
end
By default, compensations run after a workflow is cancelled or discarded. They run as a separate but linked workflow using standard Oban jobs, with all the normal recovery semantics. As with job execution, compensations are at-least-once, but they must be idempotent.
🛻 Cursor-Based Backfills
The new Oban.Pro.Backfill module works through datasets of any size to retroactively fill in
missing data, reprocess records, or apply corrections. A backfill is composed of a chain of small
jobs, where each operates on a paginated slice of records. The design avoids long-running
migrations or large transactions that can lock up a database.
Create and customize a backfill module with the Oban.Pro.Backfill behaviour. It can walk through
any ordered table (bigint and uuid alike), in ascending or descending order, and with optional
throttling to minimize load on the database or external services.
defmodule MyApp.UserBackfill do
use Oban.Pro.Backfill, queue: :backfills, limit: 1_000
import Ecto.Query
@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
The pagination-aware Cursor helpers cover common backfill tasks such as bulk database changes,
iterating over individual rows, and processing rows as a batch; like so:
@impl Oban.Pro.Backfill
def backfill(cursor, _extra) do
with {users, next} <- Cursor.fetch(MyApp.User, cursor) do
MyApp.CRM.sync_all(users)
{:cont, next, length(users)}
end
end
Once defined, you can start a backfill like any other job, or directly from within an Ecto migration:
Oban.Pro.Backfill.start(MyApp.UserBackfill)
Calling Backfill.start/1 in a migration doesn't require the application to be started, and it's
context-aware: your database won't get polluted with backfill jobs while testing.
⚙️ Streamlined Configuration
Pro's engine and core services now use shorter, slightly less bombastic, flatter module names. There's only one engine, and those are the definitive Pro implementations of each service, so simplify the naming:
Oban.Pro.Engines.SmartbecomesOban.Pro.Engine.Oban.Pro.Plugins.DynamicCronbecomesOban.Pro.Cron.Oban.Pro.Plugins.DynamicLifelinebecomesOban.Pro.Lifeline.Oban.Pro.Plugins.DynamicPrunerbecomesOban.Pro.Pruner.Oban.Pro.Plugins.DynamicQueuesbecomesOban.Pro.Queues.
These renames match similar changes made in Oban v2.24. You can also move configuration out of the
generic :plugins list and utilize dedicated configuration keys:
config :my_app, Oban,
engine: Oban.Pro.Engine,
cron: {Oban.Pro.Cron, crontab: [...]},
lifeline: Oban.Pro.Lifeline,
pruner: Oban.Pro.Pruner,
queues: {Oban.Pro.Queues, queues: [default: 10]}
For backward compatibility, the previous modules remain as (deprecated) shims and old configuration stays entirely valid, so you can adopt the new config style incrementally.
Global Worker Defaults
Also in the vein of simplified configuration, global defaults can be configured for select job options, recorded output and storage, and the new encrypted argument keyrings.
config :oban_pro, Oban.Pro.Worker,
max_attempts: 5,
priority: 1,
queue: :default,
recorded: [storage: MyApp.S3, limit: {32, :mb}]
These are defaults only. Worker options declared with use Oban.Pro.Worker or passed to new/2
take precedence, while options such as recorded and encrypted merge with the global defaults.
Defaults are validated at compile time, and configuration changes take effect when workers are recompiled.
⏰ Inline Cron Scheduling
Pro workers can now declare cron schedules inline, along with the actual code being scheduled,
using the new :cron option:
defmodule MyApp.NightlyCleanup do
use Oban.Pro.Worker, cron: "0 3 * * *"
...
end
Inline cron can also be applied to decorated functions:
use Oban.Pro.Decorator
@job cron: "@daily"
def rollup do
...
end
It's also possible to pass advanced options like timezone, guaranteed, paused, or a custom
name using the keyword version:
use Oban.Pro.Worker,
cron: [expression: "0 3 * * *", name: "custom-name", guaranteed: true],
...
Annotated workers and decorated jobs are collected at compile-time via a custom compiler, then
discovered by Oban.Pro.Cron on boot. This avoids relying on loaded modules, which is especially
important in development or test where lazy loading alters the load order.
☁️ External Recorded Storage
Recorded job output doesn't have to live on the jobs table anymore. A new Oban.Pro.Storage
behaviour lets applications offload recorded data to external storage such as S3, Tigris, Redis,
etc.
Storage can be configured for an individual worker, with localized options:
use Oban.Pro.Worker,
recorded: [storage: {MyApp.S3, bucket: "job-output"}, limit: {32, :mb}]
Or at the workflow level, as an option to Workflow.new/1:
Oban.Pro.Workflow.new(storage: {MyApp.S3, bucket: "workflow-output"})
And finally, as a global default for all recorded workers, including workflow cascades:
config :oban_pro, Oban.Pro.Worker,
recorded: [storage: MyApp.S3, limit: {32, :mb}]
Stored output is fetched transparently by Worker and Workflow functions. Then, when jobs are
pruned, the Oban.Pro.Pruner asks the storage backend to automatically delete stored data (though
lifecycle rules and TTLs are recommended).
Beyond the storage extension, recorded metadata now includes the encoded payload size, and
fetch_recorded/2 can retrieve output without the original module or decoding anything. All of
which are improvements that allow Oban Web to inspect arbitrary recorded jobs and warn before
downloading massive objects.
🧹 Persistent Pruning and Archiving
Oban.Pro.Pruner gained persisted, runtime-editable, rules to control job retention. Rules can
match any combination of queue, worker, and state, and retain jobs by age or count:
pruner: {
Oban.Pro.Pruner,
rules: [
[name: "events", queue: :events, max_age: {10, :minutes}],
[name: "discarded", state: :discarded, max_age: {7, :days}],
[name: "audit", worker: MyApp.AuditWorker, max_len: :infinity]
]
}
Rules have explicit ordering, so the sequence is fully deterministic - much better than relying on
the limited and implicit ordering of overrides. A final default rule handles jobs that no
earlier rule matched.
Because rules are persisted, they can be inserted, modified, or even paused at runtime. Like the
Cron and Queues modules, there are two sync_mode options to either preserve dynamic rules, or
treat application configuration as the source of truth.
Automatic Archiving
The final Pruner addition is declarative job archiving. Rules can be configured with archive: true to keep jobs in an archive table rather than deleting them:
[name: "audit", worker: MyApp.AuditWorker, max_age: {1, :week}, archive: true]
Jobs pruned the rule are copied in full to an oban_jobs_archive table prior to deletion.
The copy and delete operations happen in a single transaction, so jobs can't be lost between the
two tables.
Archived rows can be queried through Oban.Pro.Archive, and they load as vanilla Oban.Job structs:
[worker: MyApp.AuditWorker, state: "completed"]
|> Oban.Pro.Archive.query()
|> Oban.all_jobs()