Batches link the execution of many jobs as a group and runs optional callbacks after jobs are processed. This allows your application to coordinate the execution of any number of jobs in parallel.

Usage

Batches are composed of one or more Pro workers that are linked with a shared batch_id and optional callbacks. As a simple example, let's define a worker that delivers daily emails:

defmodule MyApp.EmailBatch do
  use Oban.Pro.Worker, queue: :mailers

  @behaviour Oban.Pro.Batch

  @impl Oban.Pro.Worker
  def process(%Job{args: %{"email" => email}}) do
    MyApp.Mailer.daily_update(email)
  end

  @impl Oban.Pro.Batch
  def batch_completed(_job) do
    Logger.info("BATCH COMPLETE")

    :ok
  end
end

Now, create a batch with new/1 by passing a list of job changesets:

emails
|> Enum.map(&MyApp.EmailWorker.new(%{email: &1}))
|> Batch.new()
|> Oban.insert_all()

After all jobs in the batch are completed, the batch_completed/1 callback will be triggered.

Handler Callbacks

After jobs in the batch are processed, a callback job may be inserted. There are four optional batch handler callbacks that a worker may define:

Each callback runs in a separate, isolated job, so it may be retried or discarded like any other job. The callback function receives an Oban.Job struct with the batch_id in meta and should return :ok (or another valid Worker.result()).

Here we'll implement each of the optional handler callbacks and have them print out the batch status along with the batch_id:

defmodule MyApp.BatchWorker do
  use Oban.Pro.Worker

  @behaviour Oban.Pro.Batch

  @impl Oban.Pro.Worker
  def process(_job), do: :ok

  @impl Oban.Pro.Batch
  def batch_attempted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:attempted,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_cancelled(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:cancelled,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_completed(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:completed,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_discarded(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:discarded,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_exhausted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:exhausted,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_retryable(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:retryable,  batch_id})
    :ok
  end
end

Hybrid Batches

Batches may also be built from a variety of workers, though you must provide an explicit worker for callbacks:

mail_jobs = Enum.map(mail_args, &MyApp.MailWorker.new/1)
push_jobs = Enum.map(push_args, &MyApp.PushWorker.new/1)

[callback_worker: MyApp.CallbackWorker]
|> Batch.new()
|> Batch.add(mail_jobs)
|> Batch.add(push_jobs)

Without an explicit callback_worker, any worker in the batch may be used for callback. That makes callback handling unpredictable and unexpected.

Customizing Batches

Batches accept a variety of options for grouping and callback customization.

Batch ID

By default a batch_id is generated as a time-ordered random UUIDv7. UUIDs are more than sufficient to ensure that batches are unique between workers and nodes for any period. However, if you require control, you can override batch_id generation at the worker level or pass a value directly to the new/2 function.

Batch.new(batch_id: "custom-batch-id")

Batch Name

Batches accept an optional name to describe the purpose of the batch, beyond the generated id or individual jobs in it. While the batch_id must be unique, the batch_name doesn't, so it can be used as a general purpose label.

Batch.new(batch_name: "nightly-etl")

Callback Workers

For some batches, notably those with heterogeneous jobs, it's necessary to specify a different worker for callbacks. That is easily accomplished by passing the :callback_worker option to new/2:

Batch.new(callback_worker: MyCallbackWorker)

The callback worker must be an Oban.Pro.Worker that defines one or more of the batch callback handlers.

Callback Options

By default, callback jobs have an empty args map and inherit other options from batch jobs. With callback_opts, you can set standard job options for batch callback jobs (only args, max_attempts, meta, priority, queue, and tags are allowed).

For example, here we're passing a webhook URLs as args:

Batch.new(callback_opts: [args: %{webhook: "https://web.hook"}])

Here, we change the callback queue and knock the priority down:

Batch.new(callback_opts: [queue: :callbacks, priority: 9])

Be careful to minimize callback_opts as they are stored in each batch job's meta.

Fetching Batch Jobs

To pull more context from batch jobs, it's possible to load all jobs from the batch with all_jobs/2 and stream_jobs/2. The functions take a single batch job and returns a list or stream of all non-callback jobs in the batch, which you can then operate on with Enum or Stream functions.

As an example, imagine you have a batch that ran for a few thousand accounts and you'd like to notify admins that the batch is complete.

defmodule MyApp.BatchWorker do
  use Oban.Pro.Worker

  @behaviour Oban.Pro.Batch

  @impl Oban.Pro.Batch
  def batch_completed(%Job{} = job) do
    {:ok, account_ids} =
        job
        |> Oban.Pro.Batch.all_jobs()
        |> Enum.map(& &1.args["account_id"])

    account_ids
    |> MyApp.Accounts.all()
    |> MyApp.Mailer.notify_admins_about_batch()
  end

Fetching a thousand jobs may be alright, but for larger batches you don't want to load that much data into memory. Instead, you can use stream_jobs to iterate through them lazily:

{:ok, account_ids} =
  MyApp.Repo.transaction(fn ->
    job
    |> Batch.stream_jobs()
    |> Stream.map(& &1.args["account_id"])
    |> Enum.to_list()
  end)

Streaming is provided by Ecto's Repo.stream, and it must take place within a transaction. While it may be overkill for small batches, for batches with tens or hundreds of thousands of jobs, it will prevent massive memory spikes or the database grinding to a halt.