Please pardon the slow reply, I’ve been on vacation for the past few weeks

This is entirely doable with a single worker through the use of :snooze. Here’s a pseudo example:

defmodule MyApp.SequentialWorker do
  use Oban.Worker, queue: :some_queue

  import Ecto.Query, only: [where: 3]

  @impl Worker
  def perform(%Job{args: args}) do
    if executable_for_user?(args) do
      # do normal stuff
    else
      # snooze for 60 seconds, make it as short as necessary
      {:snooze, 60}
    end
  end

  defp executable_for_user?(%{"user_id" => user_id}) do
    Job
    |> where(j, j.worker == Worker.to_string(__MODULE__))
    |> where(j, j.queue == "some_queue")
    |> where(j, j.state == "executing")
    |> where(j, fragment("? @> ?", j.args, ^%{user_id: user_id})
    |> MyApp.Repo.exists?()
  end
end

By checking whether there is another job executing you can force global concurrency. In this case the worker checks for other executing workers with the same user_id, though that is arbitrary and you could scope to anything you like.

It isn’t as optimized as running multiple queues with global concurrency, but it will work right now and is accurate.

Partitioned rate-limiting landed in Pro Log in | Hex. Thanks for all the support!

Working on it now that some major is finished. Of course, there’ve been many changes behind the scenes to improve queue option validation and serialization, but the groundwork is there.

As soon as something is ready for testing, I’ll announce it here