Can you elaborate on the concrete use case? I guess I just don’t see the point of something like this yet. Other languages reach for jobs in situations like this to avoid blocking, but blocking isn’t a danger in Elixir. What scenario needs Oban to execute a job, while blocking the caller of the job, which couldn’t be done by just doing the work synchronously in the caller process without Oban at all?
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
chulkilee
Serve synchronous external api (e.g. http api) for jobs performed outside of the app asynchronously such as via message queue (Kafka) or background job (oban).
Of course, I know it’s better to have external api capable of async operation (like returning 202 and provide another api to check the status… or websocket / graphql for push) - but that’s not always an option in the real world.
1
sorentwo
Oban Core Team
I have a few concrete real-world examples:
- The system runs on heterogeneous nodes where some only process media and do exports, some handle web requests, that kind of thing. Clients are connected to channels on the web node and need to be notified when work is finished on a worker node.
- Jobs in a workflow depend on work done in upstream jobs, like stages of processing. When a downstream job runs it waits for an upstream job to finish.
In both cases there are workarounds available; namely polling and using some outside pubsub mechanism. I’d much prefer to write something like this in my application:
%{some: "args"}
|> MyApp.Worker.new()
|> Oban.insert!()
|> Oban.await(15_000)
Where await/2 would return the value in a success tuple when it finishes processing.
0
benwilson512
Author of Craft GraphQL APIs in Elixir with Absinthe
@chulkilee @sorentwo Those are completely sensible use cases, but I suppose the bit I was questioning was whether a first class bespoke solution is better in the long run vs finding a way to utilize existing community tools. Conceptually, the client isn’t waiting on a job to finish, the client is waiting on a specific domain related thing to happen. By publishing, you gain complete control over the value sent out, you can even send out multiple events within the same job as various sub parts complete. All of these are completely orthogonal to the fact that the domain related activity is occurring within a job. Coupling those two feels problematic to me.
I also have library scope concerns. Obviously it’s @sorentwo’s library and he can do with it he wants, but I worry about scope creep when I see the library looking to duplicate work already found within popular and well tested libraries within the community. I just know that the moment you deliver this feature, folks will want to be able to easily add job identities, and then await on the status of those jobs on reconnect to so that you could populate eg a lists of image resize operations that are in progress. That kind of information though is extremely domain specific, you don’t want to push those attributes into the job row.
Perhaps there is a middle ground here. Maybe in lieu of building a built in pubsub, Oban could provide some useful hooks or easy to follow guide on broadcasting stuff from within a job to interested subscribers? If the concern is that people don’t want to add Erlang distribution, maybe the missing piece is a postgres based Phoenix PubSub implementation? I guess I’m pushing for a more modular solution here.
3
sorentwo
Oban Core Team
Thank you for the well reasoned and insightful response. You bring up some great points and I’m inclined to agree, I have similar concerns.
This is entirely true. While some type of await is possible you couldn’t do yield because the value isn’t stored anywhere—and it absolutely shouldn’t be.
This seems like the correct approach. There is already a lot of pubsub/notification handling in the library but it is private and lacks a few conveniences. Making that public with the addition of a few convenience functions would make all of these async workflow use-cases possible.
With a minor code change and some documentation this would be easy to do within your own app:
defmodule MyApp.ObanUtils do
import Oban.Notifier, only: [signal: 0]
alias Oban.Notifier
def insert_await(changeset, timeout \\ 5_000) do
Notifier.listen([:signal])
job = Oban.insert!(changeset)
receive do
{:notification, signal(), %{"action" => "complete", "job_id" => ^job.id} = payload} ->
{:ok, payload["result"]}
after
timeout ->
{:error, :timeout, job}
end
end
def notify(job, result) do
Notifier.notify(signal(), %{action: :complete, job_id: job.id, result: result})
end
end
defmodule MyApp.Worker do
use Oban.Worker
alias MyApp.ObanUtils
@impl Oban.Worker
def perform(_args, job) do
result = do_some_work()
ObanUtils.notify(job, result)
:ok
end
end
You’re then explicitly controlling when and how you broadcast and you’re aware of the values that are being returned.
4
chulkilee
I agree with @benwilson512 and @sorentwo in that oban’s responsibility should be well defined.
For interfaces - I’ve seen different patterns - and what’s your thoughts? Do we have established convention in elixir libraries?
- app code calls a oban function with passing callback function (like some erlang funcs)
- app defines pre/post hook (like Plug / Tesla style) - which I don’t think exists
- app uses basic per-process message (as shown in the example - using
Notifier.list/1andreceive - app subscribes telemetry: probably it doesn’t work if a job is performed by another node
- app runs an additional supervisor provided by oban, with passing custom module with
handle_event(Postgrex.Notifications)
Perhaps there is a middle ground here. Maybe in lieu of building a built in pubsub, Oban could provide some useful hooks or easy to follow guide on broadcasting stuff from within a job to interested subscribers? If the concern is that people don’t want to add Erlang distribution, maybe the missing piece is a postgres based Phoenix PubSub implementation? I guess I’m pushing for a more modular solution here.
Some notes
- Unless it’s required, it shouldn’t require erlang cluster, since I don’t think we need global knowledge here - we just need a way to a process to subscribe specific messages
- How to subscribe events - I’d say start with basic ones (e.g. job id, job status, …) - but I’m wondering others experience on content based event routing (e.g. process A subscribes any Kafka messages having record_id=A) - which my team had implemented. I think this could be another general design discussion…
1
tovarchristian21
Sorry for the late response guys. The posted code is very similar to the approach I’ve used. The job that was inserted into Oban, was a function that “prepares” a gen_stage for generating some data. This function’s execution was very fast, since it only started the necessary GenServers of all the data feed, for that reason I had to incorporate some sort of blocking mechanism, so that only 1 job would be executed (with a limit of 1 for the queue). The reason for having a single job on the queue was basically because the app can only handle some heavy operations that consume a good amount of resources. Something as basic as a receive with a timeout was all I needed. Perhaps Oban looks like way to much to limit the amount of some word execution just to 1, but there are a bunch of benefits and value that I receive form this library that it is definitely worth having it on the codebase (for example the status checking that restarts jobs that have not been finished yet). Thank you very much to all, for sharing your opinion and been so helpful.
1
tovarchristian21
BTW, is there some sort of differences when using Oban on something like Heroku? I’m feeling like the behavior of the application is different when deployed compared to my local machine. Do dynos work as nodes for Oban? the amount of them matter ?
0
sorentwo
Oban Core Team
There aren’t any differences. A dyno is an instance of your app, otherwise known as a node. You can scale them up and down as you wish, Oban will distribute jobs between the nodes.
0
tovarchristian21
So, if I have for example 3 dynos, that means that I have 3 nodes, so in my case, instead of having a queue of limit 1, the global limit is 3?
0