Oban Documentation

Recursive jobs are a pattern where a worker enqueues a new version of itself after executing. This is useful for backfilling large datasets, especially when tasks involve external services with rate limits, heavy database pressure, or long execution times that might be interrupted by deployments.

To implement a recursive job:

1. Define multiple clauses for perform/1. One clause handles the core logic, and another handles the recursion logic.
2. In the recursion clause, check if the work was successful.
3. If successful, find the next item to process (e.g., via a database query).
4. If a next item exists, enqueue a new instance of the worker using new(scheduled_in: delay) to alleviate queue pressure.
5. If no items remain, return :ok to stop the recursion.

defmodule MyApp.Workers.TimezoneWorker do
  use Oban.Worker

  import Ecto.Query
  alias MyApp.{Repo, User}

  @backfill_delay 1

  @impl true
  def perform(%{args: %{"id" => id, "backfill" => true}}) do
    with :ok <- perform(%{args: %{"id" => id}}) do
      case fetch_next(id) do
        next_id when is_integer(next_id) ->
          %{id: next_id, backfill: true}
          |> new(scheduled_in: @backfill_delay)
          |> Oban.insert()

        nil ->
          :ok
      end
    end
  end

  def perform(%{args: %{"id" => id}}) do
    update_timezone(id)
  end

  defp fetch_next(current_id) do
    User
    |> where([u], is_nil(u.timezone))
    |> where([u], u.id > ^current_id)
    |> order_by(asc: :id)
    |> limit(1)
    |> select([u], u.id)
    |> Repo.one()
  end

  defp update_timezone(_id), do: Enum.random([:ok, {:error, :reason}])
end

When designing your queue architecture, follow these best practices:

Concurrency and Distribution

• Local Limits: Queue limits are per-node, not per-cluster. If you have a limit of 2 and run 3 nodes, you will have 6 concurrent jobs globally. Use Oban Pro's Smart Engine for true global concurrency management.
• Resource Exhaustion: Ensure your system (especially database connections) can handle the sum of all queue limits. Total concurrency = sum of all limit values across all queues.

Workload Organization

• Workload Characteristics: Group jobs by their resource needs.
  • CPU-intensive: Use dedicated queues with low concurrency.
  • I/O-bound (e.g., emails): Can typically handle higher concurrency.
  • Priority work: Use dedicated queues with higher concurrency.
• External Processes: If jobs shell out to external tools (like FFMpeg or ImageMagick), use dedicated queues with low concurrency to prevent overwhelming the host system.

Important Note

Only jobs in configured queues will execute. Any job submitted to a queue name not present in your queues configuration will remain in the database and will not be processed.

To implement a job that repeats indefinitely with a fixed interval (e.g., daily digests), use a recursive pattern within the perform/1 function. To ensure you don't schedule duplicate future jobs during retries, use pattern matching on the attempt field of the Oban.Job struct.

By matching only on attempt: 1, you can schedule the next occurrence of the job before executing the business logic. Subsequent retries will fall through to a different clause that only executes the business logic without rescheduling, providing at-most-once semantics for scheduling and at-least-once semantics for delivery.

defmodule MyApp.Workers.ScheduledWorker do
  use Oban.Worker, queue: :scheduled, max_attempts: 10

  alias MyApp.Mailer

  @one_day 60 * 60 * 24

  @impl true
  def perform(%{args: %{"email" => email} = args, attempt: 1}) do
    # Schedule the next occurrence only on the first attempt
    args
    |> new(scheduled_in: @one_day)
    |> Oban.insert!()

    Mailer.deliver_email(email)
  end

  def perform(%{args: %{"email" => email}}) do
    # Subsequent retries only attempt the business logic
    Mailer.deliver_email(email)
  end
end