Using Oban on functions that return

tovarchristian21

Hello guys, I’ve been checking out Oban for implementing some processing queues for my application. Something I’ve been wondering is that enqueueing does not allow to use the return of whatever business logic you are using on the worker module. For example, when the following code is executed:

%{id: 1, in_the: "business", of_doing: "business"}
|> MyApp.Business.new()
|> Oban.insert()

The perform/2 function from the MyApp.Bussiness module will return something I need, however the insertion from the pipeline will return the Oban Structure, with some arguments and metadata that I do not need for the moment. Is there a way for enqueueing and still being able to use the return from the worker module when executing the perform/2 function ?

21 comments

/oban

0 4726 21

2020-02-26 03:53:59 UTC

Showing Posts 1 to 10

Show Best Posts
Show All (oldest first)
Show All (newest first)

NobbZ

No.

In general what you want is not possible in asynchronous background processors, by design.

The job does not need to run immediately. it can happen in a couple of seconds or take minutes or even hours before the job actually gets run.

A common approach is to let the job write its result into the filesystem or a database. You might need to also put additional information about current state into your FS/DB.

Post #1

benwilson512

Author of Craft GraphQL APIs in Elixir with Absinthe

@tovarchristian21 if you can block until the stuff is done, can you just ditch oban entirely and simply call whatever is inside perform/2?

Post #2

tovarchristian21

Makes a lot of sense to me @benwilson512 , since I’m using Oban for a single queue with a limit of 1 process, basically I just want to execute whatever is inside perform/2 ,one at a time. Do you know how to achieve something like that in Elixir, to block some process until some sort of message is received, or until some other functions ends.

Post #3

al2o3cr

If your goal is to ensure that exactly one process can be executing the code inside perform/2 at a time, what about wrapping that logic in a GenServer? You’d use GenServer.call to send a request, and the caller would block until the GenServer replied with the result. In that scenario, you’re using the GenServer’s process inbox as a work queue.

Post #4

benwilson512

Author of Craft GraphQL APIs in Elixir with Absinthe

If the caller crashes or stops, what should happen to the job?

Post #5

NobbZ

A GenServer is often used to serialize in the way you describe. Though the problem again with this is, in theory work can pile up because of the serialisation until the point that it does not finish within the default timeout of 5 seconds anymore, your caller will then crash.

Is it worth to wait potentially long for the result? Or what is the reason you want to serialize those perform calls? Quite often serialisation of the whole call is n ot necessary but only a small section of it, or even nothing.

Make sure that it is really necessary to serialise before trying. In many cases we have decided for erlang/elixir because we want to avoid serialisation…

Post #6

ityonemo

Task.async/await?

Post #7

sorentwo

Oban Core Team

On the surface this doesn’t seem like a good use of Oban. It’s hard to say exactly what you need without more details, as others have noted in this discussion.

If the goal is to do work within the same node in a blocking manner then you can just call a function directly and wait for the result.
If the work is stateful or particularly expensive then you need to introduce a bottleneck or rate limit it by putting it in a GenServer. Using GenServer.call as @al2o3cr suggested will ensure that only one request is handled at a time.
If the goal is to do work across many nodes with a global rate limit then Oban with a queue limit of 1 may work for you.

This all depends on what environment you’re running in (are nodes connected or are you in something like Heroku?), what type of data you are returning (a giant binary would be horrible to send via pubsub), the number of nodes you’re running (if you have exactly one node then you don’t need global synchronization), etc.

If you can state the problem you’re trying to solve, rather than the solution you currently have, then we can help more effectively.

Post #8

chulkilee

I have similar cases like this

Use async job queue for scaling out worker (may be in different node)
Make the async job transparent to the caller
- Making the caller a process and letting it receive a message is not an option

I’m thinking of having a genserver process, which does followings when start: (e.g. using continue)

Place async job and keep the job ref
Poll or subscribe job status change

And run the genserver under dynamic supervisor and call GenServer.call on it… so that caller sees it as blocking call.

Questions

Any feedback on the plan?
@sorentwo Is there a way to subscribe job status changes via oban? I’ve done this with postgrex notification - do you think it would be good addition to oban? If so I’ll make an issue on github

Post #9

sorentwo

Oban Core Team

There are internal mechanisms for doing this, but nothing public and no blessed way to wait for a job to complete. This has come up a lot recently and I think there is a case for a built-in system await jobs. Please do open an issue