This question inspired me to look at using batches with Oban. I’ve finally posted a recipe on implementing batches with Oban, without any built-in support.
sorentwo
Oban Core Team
smaximov
Thanks for the article! You have a small typo in the Batching Jobs for Monitoring section:
At lease one popular background job processor calls these groups “batches”, and so we’ll adopt that term here as we build it out with Oban.
arton
@sorentwo Is there a way to define priority within a queue? For instance I’d like to be able to create large amounts of batch jobs without delaying jobs created by an end-user. So my thought was to have a numerical priority indicator I could set which would allow user jobs to jump to the front of the queue if it’s filled with batch jobs with a lower priority.
dom
[moving my comment from the other thread]
In the “batch jobs” recipe, if say the final couple jobs go through final_batch_job? at the same time, wouldn’t they both get false and complete without inserting the callback job?
sorentwo
Oban Core Team
Nicely spotted, thanks! I’ve fixed it.
No, there isn’t a way to define priorities without making changes to Oban itself. Execution order is based on when a job was inserted/scheduled, which would conflict with priorities. It may be possible, but I haven’t put any effort into it.
I recommend defining separate queues for various priorities. The queues are entirely isolated, so a backup for slowdown in one queue won’t effect the others.
Quite right! The post has been updated with a few improvements to prevent those types of race conditions:
- Put the final batch job check in a task with a slight delay
- Make the job unique for a short period to prevent duplicate callbacks (not 100%, but good enough for most usage)
- Include a check for an existing status callback within the
final_batch_job?query - Recommend a unique constraint on args for the same
batch_idandstatusif absolute uniqueness is required
zimt28
Hi, thanks for Oban! I want to build a worker that removes expires sessions from the database.
I’ve created a worker which already works fine, but I’m not sure what’s the best place to start the initial job? Also, my job schedules the next run and if I stop the application with an already scheduled job and start my workers again, I end up with multiple workers.
I’ve tried the “unique” configuration, but it doesn’t allow me to schedule a new run when the worker is already running, how can I do this?
Here’s the worker with all irrelevant code removed:
defmodule App.Workers.ClearExpiredSessions do
use Oban.Worker, queue: "system", max_attempts: 10
# I've tried `unique: [fields: [:queue, :worker]]` here, but cannot schedule jobs then
@one_minute 60
# What's the best place to call this?
def start do
job = new(%{})
Oban.insert(job)
end
@impl Oban.Worker
def perform(_params, %{attempt: 1} = _job) do
delete_expired_sessions()
job = new(%{}, schedule_in: @one_minute)
Oban.insert(job)
end
def perform(_params, _job) do
delete_expired_sessions()
end
end
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
This sort of sounds like something better suited for Quantum. Quantum is built around executing things at regular intervals, where each execution should be run basically once. You don’t need a queue or bounded worker pool for that.
zimt28
Yes, I also have a lot of other regular workers using Oban, so I thought that introducing another library for just one task might be overkill.
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
A simple genserver with a 60 second Process.send_after would work too if you want to avoid another library.
zimt28
That’s my current implementation. I like Oban’s logging and retry functionalities, so I’m just playing around with it. You might be right that it’s not the best solution, but I wanted to know if it’s possible 