or “named” jobs eg “send_email_for_bill_id:777” which of course should balk at duplicates
(but haven’t thoroughly checked Oban out yet..)
outlog
0
jaimeiniesta
That can be useful as well, but of course the API needs to be flexible - in many cases it’s totally fine to repeat past jobs with the same arguments, in other cases it may be fine to repeat a job that has already finished, in cases like sending emails for bills it’s probably not… In my case I only want to check that a job is not already waiting at the queue.
0
sorentwo
Oban Core Team
This can definitely be achieved in Oban. Unlike Exq (or most any of the other libraries) you have complete control over how your jobs are inserted into the database.
First, a quick definition of “unique jobs” because in my experience the meaning can be confusing. Usually the uniqueness only applies to jobs that are in the queue. That means you can prevent putting two jobs with the same arguments in the queue at the same time, but it doesn’t prevent multiple jobs with the same arguments within a window of time. You mentioned this difference below, but I wanted to call it out for anybody that may not be familiar with the feature.
Here are some ways I can think of to implement unique jobs with different types of guarantees.
1. Partial Index
Add an index for the worker that you want to enforce uniqueness for:
create index(
:oban_jobs,
[:worker, :args],
unique: true,
where: "worker = 'MyApp.Worker' and state in ('available', 'scheduled')"
)
Then, pass on_conflict: :nothing as options to Repo.insert/2 and it will ensure you don’t have duplicate jobs.
2. Insert Helper
Create a helper inside your Repo that will check for existing jobs before trying to insert:
def insert_unique(changeset, opts \\ []) do
worker = get_change(changeset, :worker)
case get_by(Oban.Job, worker: worker, state: "scheduled") do
nil ->
insert(changeset, opts)
_job ->
{:ignored, changeset}
end
end
That function is a bit rough, but you get the idea.
Neither of these are as convenient as declaring it in the worker, i.e. use Oban.Worker, unique_for: :timer.minutes(1). I’ll think about this some more!
Honestly, I hadn’t considered first class tags before. This can be accomplished by providing additional fields to args.
Here we are adding an additional tag value to the args:
%{id: record.id, tag: "dependent"}
|> Oban.Job.new()
|> MyApp.Repo.insert()
Later, if the record is deleted we can also delete any dependent jobs within the same transaction:
dependents =
Oban.Job
|> where([j]. j.state in ["available", "scheduled"])
|> where([j], fragment("?->>'tag' = 'dependent' and ?->>'id' = ?::text", j.args, ^record.id)
Ecto.Multi.new()
|> Ecto.Multi.delete(:delete, record)
|> Ecto.Multi.delete_all(:jobs, dependents)
|> MyApp.Repo.transaction()
I hope those ideas are somewhat helpful. Thanks for checking out Oban 
7
sorentwo
Oban Core Team
Oban v0.4.0 has been released. As with the previous release, there are a bunch of fixes and added features that came from this thread and issues on GitHub. Thanks again 
Note that there are some breaking changes included in this release, but they are all minor changes that only take a find-and-replace to fix:
Directly from the CHANGELOG:
Added
-
[Oban] Add
Oban.drain_queue/1to help with integration testing. Draining a queue synchronously executes all available jobs in the queue from within the calling process. This avoids any sandbox based database connection issues and prevents race conditions from asynchronous processing. Thanks to @josevalim -
[Oban.Worker] Add
backoff/1callback and switch to exponential backoff with a base value as the default. This allows custom backoff timing for individual workers. -
[Oban.Telemetry] Added a new module to wrap a default handler for structured JSON logging. The log handler is attached by calling
Oban.Telemetry.attach_default_logger/0somewhere in your application code. (Inspired by a similar feature in Redix) -
[Oban.Queue.Producer] Guard against Postgrex errors in all producer queries using a circuit breaker. Failing queries will no longer crash the producer. Instead, the failure will be logged as an error and it will trip the producer’s circuit breaker. All subsequent queries will be skipped until the breaker is enabled again approximately a minute later.
This feature simplifies the deployment process by allowing the application to boot and stay up while Oban migrations are made. After migrations have finished each queue producer will resume making queries. Thanks to @philss. Side note, the solution was inspired by a discussion between @keathley and @adkron on the most recent Elixir Outlaws.
Changed
-
[Oban] Telemetry events now report timing as
%{duration: duration}instead of%{timing: timing}. This aligns with thetelemetrystandard of usingdurationfor the time to execute something. -
[Oban] Telemetry events are now broken into
successandfailureat the event level, rather than being labeled in the metadata. The full event names are now[:oban, :success]and[:oban, :failure]. -
[Oban.Job] Rename
scheduled_intoschedule_infor readability and consistency. Both theObandocs and README showedschedule_in, which reads more clearly thanscheduled_in. Thanks @lukerollans -
[Oban.Pruner] Pruning no longer happens immediately on startup and may be configured through the
:prune_intervaloption. The default prune interval is still one minute. Again, thanks @philss
Fixed
- [Oban.Migrations] Make partial migrations more resilient by guarding against missing versions and using idempotent statements.
10
adkron
Thanks for putting this out there. It is nice to know that our talks have an impact outside those of us that sit on the show. I’m delighted that this talk is one that people are seizing. This idea came from this episode Elixir Outlaws Episode 42: Carriage Return Line Feed?
Do you have a link to the commits that actually implemented the change? I’d love to take a look.
6
adkron
I think I found the commit. Guard against Postgrex errors in queue producers · oban-bg/oban@876c4ea · GitHub
1
sorentwo
Oban Core Team
Absolutely, thanks for putting the show out there.
I had been weighing how to properly handle the original issue (application crashing while deploying migrations because the table wasn’t available). The discussion around what is expected failure and what can be considered a “responsive” system came at the perfect time.
That’s the one 
. The change itself was rather minor and focused on a specific set of expected failures—if the oban_jobs table doesn’t exist or has structural inconsistencies it shouldn’t crash the rest of the application.
2
adkron
That is an excellent and simple solution. I especially like that you logged the information to make sure that your system is observable. You also had a test which, of course, makes me smile.
2
gdub01
I’m wondering about using oban to organize jobs that use ffmpeg.
It looks like the dampener feature you mentioned would work well with ffmpeg since it goes off mem/cpu instead of delegating to the beam - did I understand that correctly? If so, awesome =)
My main question is around > 1 apps accessing jobs… I’d like the main app to queue a job, but I want a separate app to run the job.
So in the main app:
Oban.Job.new(queue: :default, worker: :"SeparateApp.Worker")
Where SeparateApp.Worker doesn’t actually exist in the main app.. Then I actually define the SeparateApp.Worker in the separate app.
Is that a bad way of doing it? I was just thinking ffmpeg has different resource needs scales differently.
Alternatively, I could both queue the job and run the job in the separate app via api, and I believe still get job visibility (when your UI comes out) in the main app (providing it uses the same db). So it would be a separate elixir/phoenix app running in docker exposing its own api.
I had also considered just using aws here. So aws s3 adds a job to aws batch which calls a bash script in a docker container with ffmpeg. I’m guessing this would be more efficient computing wise, but it would lose the visibility and portability Oban provides.
Curious on any thoughts!
1
sorentwo
Oban Core Team
That feature isn’t quite baked yet, it definitely won’t be ready for a while. The version I had been envisioning would track reductions/memory usage from executing processes to automatically scale queues up and down based on workload. Tying that to OS processes would be significantly more difficult, especially for something like FFMpeg which is multi-threaded itself and could use all available cores even from a single job.
You can achieve this with Oban quite easily. Remember, only the exact queues passed to the Oban supervisor will be started and there isn’t any global configuration. You can use SeparateApp.Worker.new to enqueue the job from your main app and so long as the main app isn’t running that worker’s queue it will be ignored.
Here’s a more concrete example. First, define the worker with a queue other than :default:
defmodule SeparateApp.Worker do
use Oban.Worker, queue: :video_processing
@impl true
def perform(args) do
# process with ffmpeg
end
end
Then, configure your apps to start only the queues they should be processing:
# In the supervisor of your primary app
{Oban, repo: Repo, queues: [default: 20]}
# In the supervisor of your secondary app
{Oban, repo: Repo, queues: [video_processing: 5]}
Within your primary application code you can conveniently call SeparateApp.Worker.new(%{}). You are welcome to use the Oban.Job.new/2 variant you listed above, but you don’t need to.
IMO that is a great way of doing it. We do all of our media processing in a separate container for this exact reason. Resource usage is unpredictable and the BEAM can get so starved for CPU that it can’t cope the way it usually does.
Right, you would lose all of the execution guarantees and retry behavior. You can definitely make this work within a single umbrella/poncho, so long as you start apps in different containers.
2