Oban Queue Circuit Breaker

queue_breaker.ex

defmodule QueueBreaker do
  @moduledoc """
  Automatic queue pausing/resuming based on accumulated errors.
  
  ## Example
  
  Pause after 1000 errors:
  
      QueueBreaker.attach("risky-queue", 1000)
  """
  
  def attach(queue, limit) do
    counter = :counters.new(1, [:write_concurrency])
    conf = {queue, limit, counter}

    :telemetry.attach("queue-break", [:oban, :job, :exception], &__MODULE__.handler/4, conf)
  end

  def handle(_, _, %{conf: conf, job: job, queue: queue}, {queue, limit, counter}) do
    :counters.add(counter, 1, 1)

    if :counters.get(counter, 1) >= limit do
      Oban.pause_queue(conf.name, queue: queue)

      30
      |> :timer.seconds()
      |> :timer.apply_after(__MODULE__, :reset, [conf, queue, counter])
    end
  end

  def handle(_events, _measure, _meta, _conf), do: :ok

  def reset(conf, queue, counter) do
    :counters.put(counter, 1, 0)

    Oban.resume_queue(conf.name, queue: queue)
  end
end