That's awesome! Sadly this particular queue is just waiting on Discord rate limits, but I'm super ex

That's awesome! Sadly this particular queue is just waiting on Discord rate limits, but I'm super excited about trying more scalable workloads on Queues

JacobOP•3/9/23, 1:20 AM

One thing I'd like to try with Queues that would benifit from horizontal scaling: I have about 3 million files in R2 that I accidently saved under the wrong paths. It would be cool to use Queues to move them all

JacobOP•3/9/23, 1:23 AM

each one would need to read the file, parse out a timestamp, and then move it

MMatt Silverlock If you want it to go faster you’ll be able to let it scale our horizontally soon...

JacobOP•3/9/23, 1:24 AM

When will that be out? I'd love to try it this weekend

JJacob When will that be out? I'd love to try it this weekend 🙂

Matt Silverlock•3/9/23, 1:25 AM

Probably next week?

- Docs PR: https://github.com/cloudflare/cloudflare-docs/pull/7932
- WIP wrangler PR you can probably check out or grab the API endpoints from: https://github.com/cloudflare/workers-sdk/pull/2859

GitHub

queues: consumer concurrency by elithrar · Pull Request #7932 · clo...

New feature: consumer concurrency.

Changelog
Limits page updated
Update /platform/configuration/ with new properties
wrangler updates merged from cloudflare/workers-sdk#2859
New /learning/con...

GitHub

Add support for Queue Consumer concurrency by jbwcloudflare · Pull ...

What this PR solves / how to test:
This PR adds support for new Queues Consumer configuration options: concurrency_enabled and max_concurrency.
Associated docs issues/PR: In progress
Author has inc...

JacobOP•3/9/23, 1:26 AM

Sweet! Looking forward to the release

JacobOP•3/9/23, 1:26 AM

Btw can a Queue send to itself or is recursion not allowed?

JacobOP•3/9/23, 1:26 AM

I guess you could always fetch to a Worker that sends back to it

JacobOP•3/9/23, 1:27 AM

Trying to think of the best way to queue up 3 million items

JJacob Btw can a Queue send to itself or is recursion not allowed?

Matt Silverlock•3/9/23, 1:27 AM

You can do that if you want. We just don’t recommend it.

Matt Silverlock•3/9/23, 1:28 AM

R2 list() -> write keys to queue in sendBatch() -> paginate -> keep going. Store page state in a KV key or D1?

JacobOP•3/9/23, 1:28 AM

Oh that could work. and then just keep hitting the endpoint until it pages through them all

Matt Silverlock•3/9/23, 1:29 AM

Call ack() when the copy succeeds or retry() to put it back on the queue

Matt Silverlock•3/9/23, 1:29 AM

In the consumer

JacobOP•3/9/23, 1:29 AM

amazingggg I can't wait

I could use https://repeat.dev on a cron for pinging the list() worker to keep it all in the cloud too

repeat.dev | homepage

Create a webhook, schedule a task, forward emails, deploy in seconds. Everything in a simple web based editor.

JacobOP•3/9/23, 1:30 AM

I'm curious to see how best to manage concurrency. Does it automatically scale down if you call retry() too much?

Matt Silverlock•3/9/23, 1:31 AM

Yeah and/or throw errors from your consumer

Matt Silverlock•3/9/23, 1:32 AM

The bottleneck is probably the copy calls vs the list

Matt Silverlock•3/9/23, 1:32 AM

Since copy will be blocking, IIRC, until the copy is done

JacobOP•3/9/23, 1:32 AM

Yep - I can do up to 8x concurrency though which should help

MMatt Silverlock Yeah and/or throw errors from your consumer

JacobOP•3/9/23, 1:33 AM

Cool - I definitely prefer to avoid erroring out the consumer whenever possible so retry() will work great

Matt Silverlock•3/9/23, 1:34 AM

Or ack() what you need and throw to force retry of anything you didn’t ack in that batch

Matt Silverlock•3/9/23, 1:34 AM

https://developers.cloudflare.com/queues/learning/batching-retries/#explicit-acknowledgement

Batching and Retries · Cloudflare Queues

When configuring a consumer Worker for a queue, you can also define how messages are batched as they are delivered.

JacobOP•3/9/23, 1:35 AM

that works too. But throwing a global error to trigger behavior feels wrong haha

Matt Silverlock•3/9/23, 1:37 AM

#alwaysbethrowin

JacobOP•3/9/23, 1:37 AM

that's nice that it behaves that way though for when you have an unexpected error

JacobOP•3/9/23, 2:59 AM

What is the max size that a backlog can grow to?

Unsmart•3/9/23, 3:08 AM

there isnt a limit only way messages get removed is being processed or being older than 4 days

Unsmart•3/9/23, 3:09 AM

oh wait jk the backlog size is 25GB

Unsmart•3/9/23, 3:09 AM

interesting thats newly added to the limits page

Unsmart•3/9/23, 3:09 AM

I think

JacobOP•3/9/23, 2:19 PM

Ahh thanks I missed that

JJacob That's awesome! Sadly this particular queue is just waiting on Discord rate limi...

Kai•3/14/23, 10:18 PM

Any chance you can share how you did this? I've still not figured out an efficient way to do something like this.

Kai•3/14/23, 10:20 PM

Also, unrelated, is there a chance we can get a sort of dynamic key as part of queues? There's still only one queue and one consumer script but batching & concurrency is done per-key?
Seems like an easy way to enable much more reuse. Creating a new queue for every log file that needs appending or for every API region etc. is annoying and feels unnecessary, they are all exact duplicates just with a different name to make them work in parallel

KKai Any chance you can share how you did this? I've still not figured out an efficie...

JacobOP•3/14/23, 10:35 PM

It’s not terribly clever… I just watch for it to return headers saying it’s out of quota and await scheduler.wait() with whatever duration it gives in another header. Also sleeps if I get back a 429 (based on the retry_after value it sends.) I do a bit of extra sleeping after like 4x 429s in a short period to be nice, but it’s not entirely necessary.
Discord has good documentation for the rate limit headers and 429 responses it sends: https://discord.com/developers/docs/topics/rate-limits

Discord Developer Portal

Discord Developer Portal — API Docs for Bots and Developers

Integrate your service with Discord — whether it's a bot or a game or whatever your wildest imagination can come up with.

Kai•3/14/23, 10:36 PM

await scheduler.wait()

this is probably the bit I'm missing? what is that? Does that sleep but isn't billable / cause timeouts?

JacobOP•3/14/23, 10:38 PM

Basically the same as the classic await new Promise(r => setTimeout(r, 2000)); AFAIK. You will get billed for that duration

JacobOP•3/14/23, 10:38 PM

Discord only ever makes me wait like 2 seconds so I haven’t run into timeouts

Kai•3/14/23, 10:39 PM

ah ok :/ I have much tighter & complex rate limits and have so far failed to implement them on top of cloudflare without also paying a lot. Like 10k+ outstanding requests but only 200 / 1h allowed & any overrun disqualifies me from getting higher limits so

JacobOP•3/14/23, 10:41 PM

Ahh yeah if it’s that tight then Queues might not work for you

Kai•3/14/23, 10:41 PM

yeah, but thanks for sharing anyways!

JacobOP•3/14/23, 10:43 PM

Maybe you could specify the file it should append to in the Queue message? Each message you send is an object, so you could send something like {file: “log1.txt”, line: “my log line 123”}

JacobOP•3/14/23, 10:43 PM

Then the consumer would check that field and append to that file

Kai•3/14/23, 10:43 PM

yeah but then the batching gets worse and worse the more files I funnel through that one

JacobOP•3/14/23, 10:44 PM

Consumer concurrency is coming soon

Kai•3/14/23, 10:44 PM

Yeah, excited for that feature, but I don't see how that interacts with this

JacobOP•3/14/23, 10:44 PM

Ohh sorry just understood what you meant

JacobOP•3/14/23, 10:45 PM

I see so a batch has a mix of messages from all the different files, so you have more writes to your log storage

JacobOP•3/14/23, 10:45 PM

Interesting use case. I wonder if there will be some kind of partitioning in the future

Kai•3/14/23, 10:45 PM

yeah and I have a bunch of essentially duplicated logic then + either have to use huge batch sizes and risk overrunning the time, or have smaller sizes, which may end up in unreasonably small write batches

Kai•3/14/23, 10:46 PM

I mean duplicating the queues has worked for me in the past but it makes the calling code more confusing then necessary, makes the dashboard worse and updating is annoying

That's awesome! Sadly this particular queue is just waiting on Discord rate limits, but I'm super ex

Similar Threads

Similar Threads

Similar Threads