Cloudflare Developers•3y ago

There's definitely a bug here - I don't see a proper subscription. Support is equipped to help dive

There's definitely a bug here - I don't see a proper subscription. Support is equipped to help dive in.

MMatt Silverlock There's definitely a bug here - I don't see a proper subscription. Support is eq...

someguy•12/23/23, 3:56 PM

Ticket ID: 3075268, thanks!

Caio Pinna•12/27/23, 2:54 PM

Any strategies to work with more concurrency? Our current open ai account supports 2M TPM but our jobs (running on queues) only can up to 60k/min at maximum concurrency

huw•12/29/23, 12:46 AM

I have a queue with multiple ‘topics’ (messages are segmented per user), and I want to run an item after all other messages for that user have been consumed. Right now I can add a ‘cleanup’ message after all of my other sendsends, and then pick the batch off the queue and run my cleanup code at the end of the batch, but if multiple batches are executing concurrently I can’t guarantee that my cleanup code will begin after all of the other messages have finished executing. Just wondering if anyone has any ideas for architecture here, or if this is something that’s just not possible without tracking & polling flags for message status.

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

Hi, I saw this in the changelog:

Developers building on Queues can now create up to 10,000 queues per account, enabling easier per-user, per-job and sharding use-cases.

In such use-case, how would bindings to a worker work?

BBram Hi, I saw this in the changelog: > Developers building on Queues can now create...

Matt SilverlockOP•1/3/24, 9:26 AM

You can reasonably bind up to ~5000 queues (or any other kind of resource) per Worker script today.

MMatt Silverlock You can reasonably bind up to ~5000 queues (or any other kind of resource) per W...

Bram•1/3/24, 9:31 AM

And you would then access it like so: env["some-queue-" + userId].send(...)env["some-queue-" + userId].send(...)?

BBram And you would then access it like so: `env["some-queue-" + userId].send(...)`?

Matt SilverlockOP•1/3/24, 9:34 AM

Yes, if that fits your use-case. Or bucket users into a smaller set of queues.

Aster•1/3/24, 9:35 AM

Hello !!
How can I track the queue size with grafana? :3

Edit: found https://developers.cloudflare.com/queues/learning/metrics/

Metrics · Cloudflare Queues

You can view the metrics for a Queue on your account via the Cloudflare dashboard . Navigate to Workers > Queues > your Queue and under the Metrics …

Bram•1/3/24, 9:36 AM

Yea ok, got it. Is partititoning eventually going to be built in or is that unlikely? Meaning you can simply set a partition key and it will take care of creating partitions (buckets) based on the configured amount of partitions where each partition can be consumed in parallel

Bram•1/3/24, 9:36 AM

Thanks for the quick responses btw!

quambo•1/3/24, 12:53 PM

Partitioning seems to be on their radar, but the focus is on throughput right now (Some updates for @Queues users:)

quambo•1/3/24, 12:57 PM

Would love to hear if there is more specific ideas/plans around partitioning from cf side, and if its somewhere upon the next features. Right now, we're using durableobjects to implement a paritioned queue, but we end up rebuilding a lot of the queues features (like retry, concurrency, batchsize, batch timeouts etc)

Qquambo Partitioning seems to be on their radar, but the focus is on throughput right no...

Bram•1/3/24, 2:54 PM

Thanks for the link, excited for HTTP Pull. Seems like partitioning is not coming anytime soon though.

Bram•1/7/24, 1:22 PM

Is there a way to derive the queue name from the binding?
I have a worker that consumes & produces to multiple queues,

message.queue

message.queue

gives the name of the queue. I would like to avoid having to pass the queue names as additional variables (next to the bindings)

BBram Is there a way to derive the queue name from the binding? I have a worker that c...

Matt SilverlockOP•1/7/24, 3:20 PM

The

MessageBatch

MessageBatch

type that contains the messages delivered to the consumer has the queue name: https://developers.cloudflare.com/queues/reference/javascript-apis/#messagebatch

Cloudflare Queues - JavaScript APIs · Cloudflare Queues

Cloudflare Queues is integrated with Cloudflare Workers. To send and receive messages, you must use a Worker.

BBram Is there a way to derive the queue name from the binding? I have a worker that c...

Matt SilverlockOP•1/7/24, 3:20 PM

MMatt Silverlock Click to see attachment

Bram•1/7/24, 3:58 PM

Hey, I saw that indeed. What I would like to do is something along the lines of:

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

Bram•1/7/24, 4:00 PM

Especially considering that every environment can have a different queue name bound. I can work around it by using a naming convention (e.g.

staging.some-queue.v1

staging.some-queue.v1

) and so on, but was wondering if we could derive the queue name from the binding directly

Ashley•1/9/24, 10:54 PM

Messages delivered to a DLQ without an active consumer will persist for four (4) days before being deleted from the queue.

Is this the same for regular queues (e.g. not DLQs, the main queues themselves), where if there is no active consumer, messages will be deleted after 4 days? I couldn't see anything in the docs clarifying this, so was curious

AAshley > Messages delivered to a DLQ without an active consumer will persist for four (...

Matt SilverlockOP•1/10/24, 1:24 AM

https://developers.cloudflare.com/queues/platform/limits/

Limits · Cloudflare Queues

Many of these limits will increase during Queues’ public beta period. Follow our changelog or join the #queues-beta channel in our Developer Discord …

Matt SilverlockOP•1/10/24, 1:24 AM

See “maximum retention period”

Ashley•1/10/24, 9:23 AM

Thank you!

Abdallah Abis•1/12/24, 7:05 PM

Hey folks,
I'm working on a high-throughput service where we need to process around 1000 requests per second through a Cloudflare Worker. The goal is to rapidly queue these requests and consume them to update a database efficiently.

I have set up a producer worker to handle incoming requests. This worker assigns each request to one of 10 queues (chosen randomly) to manage the load.
The reason for multiple queues is to circumvent the limit of 20 concurrent consumers per queue. By having 10 queues, I aim to effectively handle 200 concurrent consumers (20 per queue).
The consumer worker is tasked with processing messages from each queue and updating the database.

Despite this setup, there's a noticeable lag in processing messages to the database. When I check the queue dashboard, I find that some queues still have messages pending even several minutes after they were posted. This delay is a critical issue for us, as our operation relies on rapid data processing and updates.

Any thoughts?

Thanks in advance.

Unsmart•1/12/24, 7:07 PM

Queues is a beta service so I would recommend using another option for production critical services.

Abdallah Abis•1/12/24, 7:24 PM

is the concurrency set per queue or per consumer. like does even my thought make sense?

divby0•1/15/24, 12:50 PM

https://developers.cloudflare.com/queues/reference/local-development/#known-issues

This known issue is still holding me back so much, it's not even funny. I am already having a worker next to my cf-pages app so I can consume from a queue to then call an api route of my pages app (because only workers can consume, pages apps can't). Now this isn't even really locally usable since we can't queue and consume from different pages/worker apps. How am I supposed to just use queues with pages apps?

Ddivby0 https://developers.cloudflare.com/queues/learning/local-development/#known-issue...

divby0•1/15/24, 12:56 PM

3 months later, we still basically can't even use queues if we have a pages app (since they can't consume and local dev doesn't work like that). Is that just something cloudflare doesn't wants us to use?

James•1/15/24, 5:29 PM

There's a lot of ongoing work with Pages + Workers convergence, but it's... slow and there's not a lot of well communicated expectations. Honestly, my stack recently when using Cloudflare has been Pages for the frontend, but offloading all the actual "work" to Workers, so I can use the rest of the ecosystem much easier, use a

wrangler.toml

wrangler.toml

, etc.

Al Hill•1/16/24, 10:01 AM

Sending a message from a worker to a queue count as a subrequest?

forslunds•1/16/24, 3:26 PM

I feel you

It looks like a long wait until something better than your solution. (Myself is sticking with GCT until its ”fixed”. Works great in dev also)

JJames There's a lot of ongoing work with Pages + Workers convergence, but it's... slow...

divby0•1/17/24, 11:39 AM

I can't easily do that because I use remix and don't want to split into a real frontend + backend.

As a workaround for everyone that's interested:
I have a queueService that basically serves as an adapter in my pages app. The producing into the queue fortunately is straightforward and for consuming I've got a worker that just calls api routes from my pages app and those routes use types I have defined in the respective producers (so I always only consume whatever I produced before in the same app).

As for the local dev, the queue producing in my queueService detects if I am on prod and uses normal queue and on local dev it doesn't push into a queue but instead makes an http request to a specific route on the worker that then proxies this back to the pages app (which isn't ideal cause things don't queue up, they simply are "let through". But it's well enough for local dev in my use case)

Vikash•1/18/24, 10:14 AM

My queue messages are duplicate processed. Is there any suggestion how to prevent that?

This is my consumer code -

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

Ddivby0 3 months later, we still basically can't even use queues if we have a pages app ...

Matt SilverlockOP•1/18/24, 11:30 AM

To be clear, Queues is still in beta, and we’re working on major asks around increased throughput, message delay controls, and addressing some bugs that affect consumer processing (per threads here). Hold tight.

MMatt Silverlock To be clear, Queues is still in beta, and we’re working on major asks around inc...

Vikash•1/18/24, 11:48 AM

Can I also add few more suggestion ?

Get messages -
```
queue.getMessages()
```
```
queue.getMessages()
```
to return an array[] will helps us to find if a message already exists with retry
Get count -
```
queue.getCount()
```
```
queue.getCount()
```
to return the number of messages currently in queue to wait or move some tasks to other services.
Is retried -
```
message.retryCount()
```
```
message.retryCount()
```
to get if a message is already retried > 3 times to take action

thanks for such an amazing product

VVikash Can I also add few more suggestion ? - Get messages - `queue.getMessages()` to...

Matt SilverlockOP•1/18/24, 12:13 PM

How would the first work? What is it meant to return? A queue can have tens of millions of messages.

zegevlier•1/18/24, 12:14 PM

Also: If you want to find if a message is already in the queue, wouldn't a method for that make more sense than one to get all messages?

MMatt Silverlock How would the first work? What is it meant to return? A queue can have tens of m...

Vikash•1/18/24, 12:23 PM

A limit, offset can be used to fetch 10 or 100. I had a requirement to remove certain messages, because customer started a job and then deleted their account. So I need to find associated messages and cancel them

VVikash A limit, offset can be used to fetch 10 or 100. I had a requirement to remove ce...

Vikash•1/18/24, 12:25 PM

The list option already available on queue dashboard, I think you can just expose the same in worker API with some filter capabilities

VVikash The list option already available on queue dashboard, I think you can just expos...

Matt SilverlockOP•1/18/24, 12:33 PM

The list is more of a “peek” for the most recent messages.

It is not for finding any message in the backlog/queue - that’s a non-trivial problem. If you had to paginate through thousands or millions of messages it would not be particularly efficient (and you would pay for reads).

If you have “stale” messages it might be better to do a KV read or D1 read in your consumer and delete (ack) messages that are no longer relevant vs trying to proactively find it in the queue.

MMatt Silverlock The list is more of a “peek” for the most recent messages. It is not for findin...

Vikash•1/18/24, 12:38 PM

Make sense, so any function to acknowledge message by id, if we store the {message_id, org_id} in D1 ? Something like ackMessage(Id)’

VVikash Make sense, so any function to acknowledge message by id, if we store the {messa...

Matt SilverlockOP•1/18/24, 12:41 PM

Today, I would put the customer ID in my messages and do a D1 query to see if that customer is_deleted or not in my consumer when I read each message.

MMatt Silverlock Today, I would put the customer ID in my messages and do a D1 query to see if th...

Vikash•1/18/24, 12:45 PM

Yes, that’s our approach for now. But I felt consumer is receiving these messages unnecessary. So should be a way to acknowledge all by id/filter/namespace

VVikash Yes, that’s our approach for now. But I felt consumer is receiving these message...

Matt SilverlockOP•1/18/24, 1:16 PM

Note that if you had

ackMessages(Array<ids>)

ackMessages(Array<ids>)

you'd still have to store those IDs / run that operation somewhere, and the costs would be the same (read, delete) as consuming them. This isn't to say there isn't a way to make this more convenient, but it won't land this quarter

Rabid•1/18/24, 6:10 PM

Whenever I add a queue binding to my wrangler.toml, or via the GUI, my cron worker no longer functions

Rabid•1/18/24, 6:10 PM

Usually I can go to "quick edit" > "Schedule > "Trigger Scheduled Event"

Rabid•1/18/24, 6:11 PM

When I have a queue binding set, this just shows a loader and does nothing, as soon as I remove the binding it starts working again

ajgeiss0702•1/18/24, 10:14 PM

Does the per-queue message throughput limit not apply to batch queueing? I'm generating a few thousand messages to test it, and ran them all at once (adding them with batches of 100 each) expecting an error, but they appeared to all go through?

ajgeiss0702•1/18/24, 10:15 PM

please ping me if you reply

Tuan•1/23/24, 1:33 AM

async queue(batch: MessageBatch<Message>, env: Env): Promise<void> { batch.ackAll(); for (const message of batch.messages) { const task = message.body as unknown as Task; // d1.insert(`Running task ${task.id}`); // Heavy processing here may take upto 1 minute per batch } }

There's definitely a bug here - I don't see a proper subscription. Support is equipped to help dive

Similar Threads

Similar Threads

Similar Threads