budgie

How can I snooze a whole queue in Oban until a certain time? (for daily API quota)

I have a daily quota from an API that I’m calling in Oban jobs. When I get an error response because of the quota exceeded, I want to postpone all jobs in the queue until 12:00amPT when it resets.

I see there is the ability to postpone jobs based on a {:snooze, time} response, but is there a way to do it on a whole-queue basis? Or is the best action to just run through all the jobs and have them reschedule individually? I’m trying not to hit the API too much with requests that won’t execute.

12 comments

/oban

17 791 12

2024-11-17 21:46:16 UTC

Marked As Solved

sorentwo

Oban Core Team

This is easily accomplished with two workers and two queues.

The worker that’s interacting with your API will respond to a rate limit error by pausing the queue and scheduling a job to resume the queue at midnight.

Like this:

defmodule MyApp.APIWorker do
  use Oban.Worker, queue: :some_api

  @impl true
  def perform(%{args: args, queue: queue}) do
    case MyApp.make_api_call(args) do
      {:ok, _whatever} ->
        :ok

      {:error, :rate_limit} ->
        tomorrow =
          Date.utc_today()
          |> Date.shift(day: 1)
          |> DateTime.new!(~T[00:00:00])

        %{queue: queue}
        |> MyApp.APIMonitor.new(scheduled_at: tomorrow)
        |> Oban.insert!()

        Oban.pause_queue(queue: :some_api)

        {:snooze, 60}
    end
  end
end

The worker that resumes the queue runs in a different queue that isn’t paused, and it’s marked as unique to prevent enqueuing duplicates if multiple jobs hit the rate limit at the same time.


defmodule MyApp.APIMonitor do
  use Oban.Worker, queue: :monitors, unique: true

  @impl true
  def perform(%{args: %{queue: queue}}) do
    Oban.resume_queue(queue: queue)
  end
end

This has the advantage that it’s resilient against restarts so you can deploy in between and the queue will automatically pause itself again (or the queue will remain paused if you’re able to use DynamicQueues). It also avoids snoozing, which can still churn jobs (and mess with the attempt count if you’re not using a Pro worker).

Post #6

Also Liked

budgie

I love you! Simple and elegant, cheers. I know people have been defaulting to Oban for a while but I hadn’t gotten the chance to use it. Yesterday I started looking into the docs and watched your oban architecture video. Super impressing engineering, thanks for the great library.

Post #7

MRdotB

I have an automated YouTube channel, and all the tasks are handled using Oban. For the YouTube uploading part, I faced similar issues with the YouTube API quota limitations, which only allow a certain number of operations within a 24-hour period.

If you’re interested, I’ve also written a tutorial about uploading to YouTube with Elixir

From my experience running this pipeline for multiple years, it’s much more convenient to track how many quota units you have left on your API key so you can max out the quota. This becomes especially helpful if you plan to add more API keys either for handling actions on behalf of different users or just to increase your overall quota without doing the google verification.

I manage my multiple Google credentials and their current quotas using a model in my database:

schema "google_credentials" do
  field :name, :string
  field :client_id, :string
  field :client_secret, :string
  field :refresh_token, :string
  field :quota, :integer, default: 0

  timestamps()
end

To handle API requests, I wrap each operation with a database query to find a credential with enough quota available. If none are found, the job is rescheduled to run after the next quota reset, with the highest priority to ensure it gets handled before new jobs are queued.

Here’s a simplified example of how I manage the quota for YouTube uploads and other API operations:

def insert_simple(channel_name, video_path, title, description, tags) do
  operation_cost = 1_600

  quota_wrapper(channel_name, operation_cost, fn client ->
    Api.insert_simple(client, video_path, title, description, tags)
  end)
end

def set_thumbnail(channel_name, youtube_video_id, thumbnail_path) do
  operation_cost = 50

  quota_wrapper(channel_name, operation_cost, fn client ->
    Api.set_thumbnail(client, youtube_video_id, thumbnail_path)
  end)
end

defp quota_wrapper(channel_name, operation_cost, fun) do
  with {:ok, credential} <- Google.fetch_credential_with_quota(channel_name, operation_cost),
       client <- Api.new(credential.name),
       :ok <- Google.increment_credential_quota(credential.name, operation_cost) do
    fun.(client)
  else
    :error ->
      Logger.info("Quota increment failed")
      {:error, :quota_increment_failed}
  end
end

And then you reset the quota at 0 Los_Angeles timezone which is what’s youtube is doing

    {
      Oban.Plugins.Cron,
      crontab: [
        {"0 0 * * *", Lor.Worker.ResetCredentialsQuota}
      ],
      timezone: "America/Los_Angeles"
    }
  ],

Post #13

tfwright

Are you using a queue size in the thousands? Because it seems like that’s the only way that many requests would go through after a rate limited response came back…

I don’t have experience with pause_queue or honestly with a lot of Oban features so there might be something out there I’m not personally aware of. At first glance I’m not sure it’s a much better option since it doesn’t actually stop executing jobs so a high queue size would still be a potential problem?

One other thought, have you looked into customizing the backoff algo? That is frequently the right tool for dealing with rate limiting.

edit on second look the snooze return value doesn’t seem like a bad option, seems like the same issue would be with simultaneous execution…

Post #4