Separating job/queue/task definition/configuration from enqueuing

[victoronascimento]

First of all, thanks a lot for all your work! This is one of the best documented persistent job queues I could find in Rust. Thanks a lot!

I am still a little stuck here on the best/proper way to define/configure tasks/queues and separate that from enqueuing.

What I want to achieve is something like:

• I have 2 deployments types of my server: API and Workers (and probably others)
• I want to configure queues/jobs/tasks at boot time (before readiness is successful)
• After that, I want to enqueue jobs in the API deployment and start workers in the Worker deployment

From what I am reading the docs, I will need to keep references from the boot setup phase in a global state so that I can call the enqueue on these references in other parts of the server. On the workers side it is easier because I can start workers and just wait for the shutdown signal.

Is there a way to call enqueue without needing these references? Am I missing something here? I feel like this scenario is the typical use-case here but still I don't think I've seen an example with this kind of scenario. Sorry if I missed anything from the docs/repo.

P.S.: I am new to Rust ;) I was spoiled by the excellent Oban library in Elixir which is very easy and performant... They have some impressive architecture and work on this subject that might be interesting as a research area:

• https://getoban.pro/articles/optimizing-connections-and-bloat-in-oban-pro
• https://getoban.pro/articles/one-million-jobs-a-minute-with-oban

[maxcountryman]

Is there a way to call enqueue without needing these references?

You can enqueue manually if you like, but you'll still need to provide a Task impl with its correct associated input.

[victoronascimento]

Even with manually enqueuing I'd need a "built" queue, right?

Trying to rephrase here so that my intent is clearer: there is no "static" way of calling enqueue, right? I don't want to call build on a job/queue every time I want to enqueue jobs. Build calls the database for setting things up and so it would wasteful to do so (if I've understood properly the code). Because of that, once I call some setup function for the jobs/tasks/queues, I need to keep their references in a shared state so that I can call enqueue on them.

Just as an example, Apalis has some ways for separating things. One of them is a SQL function. This is nice as it is easy enough to be called anywhere. The other way is similar to Underway where you need to keep track of a "storage" and pass it around in a shared state. Though, currently, in Apalis the pg pool is not shared between queues/tasks. This complicates things...

Anyway, I hope this makes more sense!

[maxcountryman]

It wouldn't really be possible to do that in a type safe way, if I'm following what you're asking for.

Put differently, we intentionally don't expose a public API via Postgres (because that wouldn't enforce type safety!)

[imothee]

@maxcountryman I ran into this as well and I think the underlying question is the following.

Is there a queue-lite type that would enforce some kind of type safe api similar to the queue or job that you can just instantiate cheaply with a pgpool without having to have the core queues in context.

Ie. if I have Jobs/Queues and want to call one from an axum_handler today we need to store all queues in the context and pull them out of app_state. It's not terrible but plumbing a new queue through to an endpoint requires multiple file changes and a larger state.

So if there was a way to just tell underway "this queue exists, trust me" we wouldn't need to be redundant or hold the global references. I don't actually know the performance overhead of just recreating a one off Job so maybe it's actually not a huge issue but I don't want to assume.

Some ways we could approach this if this use case was something we wanted to support
- This could be a flag on building a job like Job::builder().pool(&pool.clone).skip_setup()
- This could be a new interface that doesn't recreate the queues at all EnqueueJob::builder().pool(&pool.clone).enqueue(&job_struct)

Either way the end goal is a job or queue that we can create on the fly rather than storing the references.

[maxcountryman]

@imothee I'm open to exploring the options if you have some ideas in mind, especially if it's additive (I don't want to break type safety for the current API).

[imothee]

I was going to see if I could come up with a solution to the global shutdown broadcast issue so I'm happy to play around with a few approaches/branches and see what might make sense :)
We've been deep diving job queues for rust and underway gets the closest to the kind of interface and guarantees/perf so I'd love to contribute back and support making it better.

[maxcountryman]

@imothee lmk if you think the PR I opened will address the shutdown problem the way you were hoping (if not, no worries, but it was easier enough to implement as one obvious solution).

[victoronascimento]

Hi @imothee !

Have you had a chance to test your designs? I am quite interested in trying it out :) Both ways seems it would work for my use-cases. Also, more people seem to be asking the same question (looking in the discussions).

I could also try and cook it up myself if time is tight on your side.

Thanks for bringing ideas to this thread!

[imothee]

hey @victoronascimento I wanted to spend some time here but we found a workaround that unblocked us to focus on some other business priorities without adding any new functions to the code.

I had a half baked pr that basically had a queuebuilder that took a "skip setup" step that threw an error if it hadn't been bootstrapped already

let queue = Queue::builder()
        .name(QUEUE_NAME)
        .init(false)
        .pool(pool.clone())
        .build()
        .await?;

I didn't get it to a point that I felt happy I had caught the edge cases so we ended up going with the multitask example (https://github.com/maxcountryman/underway/blob/main/examples/multitask/src/main.rs) - the only overhead is that we have one mega job we have to extend with every new job but it's not that much work and it lets us do error logging in a single place on the parent task.

We then have a global Arc and have it exposed via global lazy static and in the axum appstate and it's worked flawlessly for simplicity to not have to spin new ones up.