Support rate limiting globally by job, rather than globally by queue

[TkTech]

We may be able to generalize the rate limiting to make it suitable for both use cases, albeit with more setup. Something like:

queue = Queue("default", rate_limit_key="amazon_api_requests")
job = Job(...).with_rate_limit_key("amazon_api_requests")
job = Job(...).with_rate_limit_key("amazon_api_requests").with_rate_limit_partition("customer_id")

Would cause both that job and that queue to share the same global rate limit, with the 3rd job having a rate limit per distinct customer_id.

You'd then initialize/reset/adjust a global rate limit with:

await chancy.push_rate_limit("amazon_api_requests", 30, 60)

[ppsmilesi]

Hey @TkTech , thank you for this amazing library. I don't understand why it is not more promoted and known. Another feature that would be nice to set by Job instead of by Queue is concurrency.

Do you think it's OK for the moment to have one queue per Job as a workaround for now ?

[TkTech]

Hey @TkTech , thank you for this amazing library.

Thanks for giving it a try!

I don't understand why it is not more promoted and known.

I've only recently started to promote is a bit, as it's now used in enough distinct projects that I'm confident the API is usable in the majority of use cases.

Do you think it's OK for the moment to have one queue per Job as a workaround for now ?

Yep this is totally okay as long as you're not using thousands of queues. Since each Queue operates as its own task to prevent one backlogged queue from blocking another, that means each queue runs its own SELECT...FOR UPDATE to grab jobs. If you've got lots of queues all busy constantly you might want to tweak the polling interval so that they're staggered.

[ppsmilesi]

Hey @TkTech,

I look forward to this feature, supporting rate limiting globally by job, rather than globally by queue ! and having a rate limit on a partition key like customer_id in your example would be amazing. Is it a shame too few queue systems support this feature. I only know about kafka.

I would like to do it for concurrency as well, on globally by job. and having a concurrency for a given job on a partition key too.

Do you have an idea on a strategy to implement this ? I may find time to work on it.

[TkTech]

Hm.

The current system uses a separate table with each rate limit indicated by a key. Currently, that key is just the queue name. If we parse the partitioned keyword value out of the arguments and into its own column at the time we push it to the database, we could change the key to f{queue_name}_{row.partition_value}. Writing the query for this in a way that the loop always pulls up to the concurrency limit and skips jobs that are blocked by rate limits is very tricky, IMO.

The easiest is probably to fake it - when a job is pushed to a queue with a partition rate limit, we fake it and make a new queue for that value with a new setting that optionally allows the queues to auto-delete after a timeout. This would keep the existing code 99% the same, but high cardinality could result in a ridiculous number of queue consuming tasks.