Oban. Pro. Backfill behaviour
(Oban Pro v1.8.0-rc.0)
Process large data sets gradually with a chain of small, resumable jobs.
Backfills are useful for updating existing records, deleting stale data, or performing external work without a long-running migration or one large transaction. Each job processes a limited window of rows and advances a cursor until no rows remain.
Usage
Define a backfill with use Oban.Pro.Backfill and implement backfill/2. Pass the cursor to
an Oban.Pro.Backfill.Cursor helper to process the next window and continue the backfill:
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([u], is_nil(u.migrated_at))
|> Cursor.update_all(cursor, set: [migrated_at: ^DateTime.utc_now()])
end
endStart the backfill from a migration:
defmodule MyApp.Repo.Migrations.BackfillUserMigratedAt do
use Ecto.Migration
def up do
Oban.Pro.Backfill.start(MyApp.UserBackfill)
end
endThe migration inserts the first job when it runs in dev or prod. In test, start/1 skips
insertion so migration tests don't leave backfill jobs behind.
Non-Database Backfills
Not every backfill is a single statement. When each row needs individual work, such as an
external API or inference call, use Cursor.each/4:
@impl Oban.Pro.Backfill
def backfill(cursor, _extra) do
Cursor.each(MyApp.User, cursor, fn user ->
embedding = MyApp.Inference.embed(user.bio)
MyApp.Repo.update_all(where(MyApp.User, id: ^user.id), set: [embedding: embedding])
end)
endUse Cursor.fetch/3 when you need the whole window for a batch request or to process rows
concurrently with Task.async_stream/3. It returns the rows and the advanced cursor. A window
may run more than once on retry, so any external work must be safe to repeat.
Options
Set options as defaults with use Oban.Pro.Backfill, or override them with new/2 and
start/2:
:key— The unique, ordered column used to track progress. It may be an integer or UUID. Defaults to:id.:limit— The maximum number of rows per window. Defaults to5_000.:order— Process keys in:ascor:descorder. Descending order requires a key with a meaningful order, such as an integer or UUIDv7. Defaults to:asc.:throttle— Pause between windows by the given period, e.g.{1, :second}, to limit load. Defaults to no delay.:timeout— Limit how long a single window may run, in milliseconds, before Oban cancels it. Defaults to no limit.:dry_run— Process every window while rolling back its database writes. Use it to verify that a backfill terminates and estimate how many windows it requires. External side effects aren't rolled back. Defaults tofalse.
Extra Data
Pass data needed by every window as the first argument to new/2, or with the extra: option
to start/2. The data arrives as the second argument to backfill/2 and preserves Elixir
terms such as tuples, atoms, and structs.
MyApp.UserBackfill.new(%{tenant_id: 42}, limit: 500)Idempotency
A window may run more than once after an error or node shutdown. Make the work in
backfill/2 safe to repeat. Backfill doesn't wrap the callback in a transaction, though a
dry_run does use a transaction to roll back database writes.
Summary
Types
An opaque cursor. Pass it to the Oban.Pro.Backfill.Cursor helpers and advance it by
returning the value they hand back.
Return value of backfill/2.
Callbacks
Called once per iteration with the current cursor and the user-supplied extra map.
Functions
Cancel every job in a backfill, stopping the chain.
Start a backfill from an Ecto migration, a remote console, or another one-shot context.
Types
@type cursor() :: map()
An opaque cursor. Pass it to the Oban.Pro.Backfill.Cursor helpers and advance it by
returning the value they hand back.
@type result() :: :halt | {:cont, next :: cursor(), count :: non_neg_integer()} | {:error, term()} | {:cancel, term()}
Return value of backfill/2.
Callbacks
Called once per iteration with the current cursor and the user-supplied extra map.
The callback may return:
:halt— no successor is inserted and the chain stops{:cont, next, count}— a successor job is inserted with the advancednextcursor and the accumulatedcountadvanced bycount{:error, reason}— the job errors and Oban handles with standard retry{:cancel, reason}— the job is cancelled and the chain stops
The Cursor helpers conveniently return these values, so a callback is usually a single piped
expression.
Functions
@spec cancel(Oban.name(), Oban.Job.t() | String.t()) :: {:ok, non_neg_integer()}
Cancel every job in a backfill, stopping the chain.
Cancellation affects all the backfill's jobs that aren't already in a terminal state, including the one currently executing.
Pass either the backfill_id returned in a job's meta, or a Job struct from inside a running
iteration.
Examples
Cancel a backfill by id:
Oban.Pro.Backfill.cancel("some-uuid-1234-5678")Cancel the whole backfill from within a running iteration, given the executing job:
Oban.Pro.Backfill.cancel(job)Cancel a backfill with a custom Oban instance name:
Oban.Pro.Backfill.cancel(MyApp.Oban, job)
@spec start(worker :: module(), opts :: keyword()) :: :ok | {:ok, Oban.Job.t()} | {:error, Ecto.Changeset.t()}
Start a backfill from an Ecto migration, a remote console, or another one-shot context.
Inside a migration, start/2 uses the migration's repo automatically. Outside a migration, it
uses the repo configured for the :oban instance, or an explicit :repo option. A running Oban
instance isn't required when a repo is available.
In the test environment, calls from a migration return :ok without inserting a job. Calls
outside migrations insert normally.
Examples
Start a backfill from a migration:
Oban.Pro.Backfill.start(MyApp.UserBackfill, limit: 5_000)Pass extra: data through to each backfill/2 call:
Oban.Pro.Backfill.start(MyApp.UserBackfill, extra: %{tenant_id: 42})Use a named Oban instance outside a migration, such as from a remote console:
Oban.Pro.Backfill.start(MyApp.UserBackfill, oban: MyApp.Oban)Pass a repo explicitly when Oban isn't running:
Oban.Pro.Backfill.start(MyApp.UserBackfill, repo: MyApp.Repo)