Pausing Jobs Instead of Queue?

Would using the built in priority feature work here instead of trying to juggle priority yourself via pausing and unpausing?

Nope it would not, because our entities have already occupied those priorities.

For the specific part that you quoted, we are “juggling” the priority ourselves because we don’t want those entities to use Oban for the time being, since we are moving them to Oban in phases.

Basically, we have entities with our own assigned priorities from 1 to >1000, with the lower number being the highest priority. Since moving an entire system is tricky, we want to move only internal priorities 300 and above to Oban, while keeping internal priorities <300 on our current system. These 1-1000+ priorities are grouped according to Oban’s 0-9 priorities.

The Ecto.StaleEntryError you’re getting is from a bug in DynamicQueues that was fixed in the rather recent Pro v1.5.0-rc.4.

Note that Oban.check_queue doesn’t query the database. It returns information directly from the queue’s producer process.

The easiest workaround is to upgrade to the latest RC (it’s stable, no known bugs at this point) and avoid the annoying StaleEntryError bug .

The Ecto.StaleEntryError you’re getting is from a bug in DynamicQueues that was fixed in the rather recent Pro v1.5.0-rc.4.

Hi, you may remember me from this post Getting StaleEntryError on DynamicQueues.update

I have been on the latest Oban Pro rc version ever since then, but unfortunately am facing this error now.

Of course

You’re positive it’s rc.4? The issue was reproducible in tests and there were numerous logic changes, including a rescue for that specific exception.

You’re positive it’s rc.4? The issue was reproducible in tests and there were numerous logic changes, including a rescue for that specific exception.

Yes, I’m positive. In my mix.lock, the version is 1.5.0-rc.4. Unless the mix.lock isnt an accurate indicator?

No, that’s an accurate indicator. I’m not sure how you’re still seeing that error then. Which mechanism are you using to pause? Will you share your current code?

Just to remove all variables, did you do these steps?

rm -rf _build deps
mix do deps.get, compile

And then try again?

I had not, but I just tried and it didn’t resolve it unfortunately

Which mechanism are you using to pause? Will you share your current code?

Just pausing the queue based on the queue name. Sure. I have also included the other code I tried as commented out code

1st module

  defp pop_and_send(%{queue: _queue, name: _name, priority_queue: []} = state) do
    ObanQueue.use_oban?() |> maybe_resume_oban_queues(state)
  end

  # to modify when high priority entities are moved to oban
  defp pop_and_send(%{name: name, priority_queue: priority_queue} = state) do
    # IO.inspect(state.oban_paused, label: "oban_paused when priority queue not empty")

    # Pause when `priority_queue` is not empty
    if ObanQueue.use_oban?() and !state.oban_paused, do: ObanQueue.pause_queue(name)
    # if ObanQueue.use_oban?() do
    #   queue = Oban.check_queue(queue: ObanQueue.set_queue(name))
    #   if !queue.paused, do: ObanQueue.pause_queue(name)
    # end

    # ... <logic involving our existing system here>

    %__MODULE__{state | priority_queue: priority_queue, oban_paused: true}
  end

  # check if queue is paused first
  defp maybe_resume_oban_queues(true, %{name: name, oban_paused: true} = state) do
    ObanQueue.resume_queue(name)

    %__MODULE__{state | oban_paused: false}
  end

  defp maybe_resume_oban_queues(true, %{oban_paused: false} = state), do: state

  defp maybe_resume_oban_queues(false, %{queue: queue, name: name, priority_queue: []} = state) do
    case Qex.pop(queue) do
      {:empty, _queue} ->
        state

      {{:value, pid}, queue} ->
        # ... <logic involving our existing system here>

        %__MODULE__{state | queue: queue}
    end
  end

ObanQueue module

@doc "Resume queue based on SSP"
  # def resume_queue(ssp) do
  #   try do
  #     Oban.resume_queue(queue: set_queue(ssp))
  #   catch
  #     type, reason ->
  #       error = Exception.format(type, reason, __STACKTRACE__)
  #       Logger.warning("Blah blah #{error}")
  #       :ok
  #   end
  # end

  def resume_queue(ssp), do: Oban.resume_queue(queue: set_queue(ssp))

  @doc "Pause queue based on SSP"
  # def pause_queue(ssp) do
  #   try do
  #     Oban.pause_queue(queue: set_queue(ssp))
  #   catch
  #     type, reason ->
  #       error = Exception.format(type, reason, __STACKTRACE__)
  #       Logger.warning("Blah blah #{error}")
  #       :ok
  #   end
  # end

  def pause_queue(ssp), do: Oban.pause_queue(queue: set_queue(ssp))

set_queue is a function that just identifies the queue name based on the variable