Guide

Batches

Enqueue a group of jobs and run a callback when the whole batch completes.

Create and enqueue a batch#

A batch tracks a group of related jobs. Enqueue the jobs inside batch.enqueue and each is tagged with the batch id:

batch = Pgbus::Batch.new(
  on_finish: BatchFinishedJob,
  on_success: BatchSucceededJob,
  on_discard: BatchFailedJob,
  description: "Import users",
  properties: { initiated_by: current_user.id }
)

batch.enqueue do
  users.each { |user| ImportUserJob.perform_later(user.id) }
end

Callbacks#

Callback	Fired when
`on_finish`	All jobs completed (success or discard).
`on_success`	All jobs completed successfully (zero discarded).
`on_discard`	At least one job was dead-lettered.

A callback job receives the batch properties hash as its argument:

app/jobs/batch_finished_job.rb

class BatchFinishedJob < ApplicationJob
  def perform(properties)
    user = User.find(properties["initiated_by"])
    ImportMailer.complete(user).deliver_later
  end
end

How batches work#

Batch.new(...) creates a row in pgbus_batches with status: "pending".
batch.enqueue { ... } tags each enqueued job with the batch id in its payload.
As each job completes or is dead-lettered, the executor atomically updates the batch counters.
When completed + discarded == total, the status flips to "finished" and the callback jobs are enqueued.
The dispatcher cleans up finished batches older than 7 days.

The pgbus_batches table comes from the base pgbus:install migration — no separate generator is needed for batches.