is there a way with which i could limit the processing rate of a queue besides with concurrency + ba

is there a way with which i could limit the processing rate of a queue besides with concurrency + batch size? like i want to limit a certain number of messages / h

oof2win2OP•11/7/24, 7:22 PM

actually nvm i can do this with D1 and a queue table (even though kinda dumb, it works tho)

Ooof2win2 is there a way with which i could limit the processing rate of a queue besides w...

DoggeSlapper•11/7/24, 7:33 PM

u can using pull style consumer

oof2win2OP•11/7/24, 7:33 PM

i'm using workers so that doesn't really make sense here

oof2win2OP•11/7/24, 7:34 PM

easier to build my "own" queue via d1 + cron

DDoggeSlapper Queue wrangler dev concurrency is 1 right?

DoggeSlapper•11/7/24, 7:40 PM

Is there a way to run multiple consumers locally?

DoggeSlapper•11/7/24, 7:47 PM

I mean for concurrency not 2 diffrent consumers on same queue, or there is no way in local to have concurrency?

Sam•11/11/24, 2:58 PM

Is there anyway to increase concurrency beyond 250 workers?

I had 900,000 messages to queue last night and it did work but it took over 10 hours to complete all of the workload.

SSam Is there anyway to increase concurrency beyond 250 workers? I had 900,000 mess...

nclevenger•11/11/24, 3:05 PM

Multiple queues?

Ibrahim John•11/13/24, 8:41 AM

Hello guys, I there a way to know that the queue is complete

IIbrahim John Hello guys, I there a way to know that the queue is complete

Matt Silverlock•11/13/24, 11:43 AM

You can see how many messages are in the queue via the dash or the API: https://developers.cloudflare.com/queues/observability/metrics/

Cloudflare Docs

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 tab you’ll be able to view line charts describing the number of messages processed by final outcome, the number of messages in the backlog, and other important indicators.

can•11/20/24, 9:29 AM

Hello y'all, happy coding to everybody :pikahey:

I have a question regarding to the getting messages from queue. I have a producer worker which is a queue manager in general which manages routing for queues. basically, if payload

{ targetQueue: "test-runner", data: {} }

{ targetQueue: "test-runner", data: {} }

. Let's say I'm sending data to "test-runner" queue via queue manager. For some reasons, sometimes (more oftenly) my consumer worker doesn't get these messages from the queue, it just stay still. Sometimes I'm seeing

Exceeded CPU Time

Exceeded CPU Time

error and that happen right after it runs for the first time and then I can't get anything back from the consumer worker.

In my queue handler i'm doing 2 db operations and making a request to queue manager with the necessary data to keep process going.

What am I doing wrong here? Would love to get some pointers

Thanks in advance!

Ccan Hello y'all, happy coding to everybody :pikahey: I have a question regarding t...

Pranshu Maheshwari•11/20/24, 3:38 PM

It looks like your consumer Worker is running into CPU time limits (documented here)

Pranshu Maheshwari•11/20/24, 3:40 PM

You'll need to change your consumer Worker to handle the messages in under 30s of CPU time! Otherwise your Worker will time out, and the message will fail.

Pranshu Maheshwari•11/20/24, 3:45 PM

For some reasons, sometimes (more oftenly) my consumer worker doesn't get these messages from the queue, it just stay still.

What's likely happening here is that your consumer Worker is taking a long time to process messages, and is erroring out.

Consumer workers autoscale in response to load, but if your Worker is returning errors (such as

Exceeded CPU time

Exceeded CPU time

), they won't auto-scale (see here for details).

Because your consumer worker is erroring out, your consumer isn't autoscaling: it's likely stuck at a concurrency of 1. And so, it might appear that messages aren't being received, but it's more likely that your (single) consumer is stuck processing the previous message.

Cloudflare Docs

Consumer concurrency | Cloudflare Queues

Consumer concurrency allows a consumer Worker processing messages from a queue to automatically scale out horizontally to keep up with the rate that messages are being written to a queue.

pao ramen•11/20/24, 4:21 PM

What does "average backlog" mean? How to see the current backlog? I have a queue that can't stop increasing the average backlog no matter what I do, and I'm concerned. It's at concurrency 1 for some reason.

johtso•11/21/24, 3:59 PM

Is there a way to get the name of the queue from the Queue binding object?

johtso•11/21/24, 4:00 PM

my worker consumes from multiple queues and I want to switch based on the queue name, but I'd rather not hard-code it, especially as the queue has a different name depending on the environment

Jjohtso my worker consumes from multiple queues and I want to switch based on the queue ...

Brett Willis•11/21/24, 7:41 PM

On the consumer side,

batch.queue

batch.queue

is the name of the queue that the batch of messages is from:
https://developers.cloudflare.com/queues/configuration/javascript-apis/#messagebatch

Cloudflare Docs

JavaScript APIs | Cloudflare Queues

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

BBrett Willis On the consumer side, `batch.queue` is the name of the queue that the batch of m...

johtso•11/21/24, 8:15 PM

Yep, but I'm looking to do something like

batch.queue === env.FOO_QUEUE.name

batch.queue === env.FOO_QUEUE.name

johtso•11/21/24, 8:17 PM

Otherwise I'm hard coding the queue name in multiple places, and if I have multiple environments so, foo-queue-prod, foo-queue-staging.. then I have to do some

startsWith

startsWith

ugliness

Jjohtso Yep, but I'm looking to do something like `batch.queue === env.FOO_QUEUE.name`

Brett Willis•11/21/24, 8:59 PM

Ah right. Looks like the binding itself only has the

send()

send()

methods.
https://developers.cloudflare.com/queues/configuration/javascript-apis/#queue
You could define an env variable with the queue name. Because the env variable is defined right beside your queue binding in your

wrangler.toml

wrangler.toml

it wouldn't be too bad?

Cloudflare Docs

JavaScript APIs | Cloudflare Queues

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

BBrett Willis Ah right. Looks like the binding itself only has the `send()` methods. https://...

johtso•11/21/24, 9:46 PM

Oh yeah will probably end up doing that, obviously not the end of the world, but do wonder whether it would be a nice API addition

Jjohtso Is there a way to get the name of the queue from the Queue binding object?

John Spurlock•11/22/24, 3:55 PM

unfortunately not, I end up doing something like this. It would be swell if this was available as a property on the queue binding itself

Pranshu Maheshwari•11/22/24, 3:57 PM

thanks for highlighting this: let me see what we can do here!

Jeff (vec4me)•11/23/24, 1:29 AM

All of my messages in the queue are being dispersed all at once!
I want them to send one at a time, 1 every 5 seconds. Can anyone help me with this?
Here's what my consumer looks like:

export default {
  async queue(batch) {
    const ntfyUrl = 'https://ntfy.sh/doofy';

    const message = batch.messages[0];
    if (message) {
      try {
        const response = await fetch(ntfyUrl, {
          method: 'POST',
          body: message.body,
        });

        if (response.ok) message.ack();
      } catch (error) {
        console.error(`Failed to process message: ${error}`);
      }
    }
  },
};

export default {
  async queue(batch) {
    const ntfyUrl = 'https://ntfy.sh/doofy';

    const message = batch.messages[0];
    if (message) {
      try {
        const response = await fetch(ntfyUrl, {
          method: 'POST',
          body: message.body,
        });

        if (response.ok) message.ack();
      } catch (error) {
        console.error(`Failed to process message: ${error}`);
      }
    }
  },
};

Jeff (vec4me)•11/23/24, 1:31 AM

And here's what the queue's settings look like:

JJeff (vec4me)And here's what the queue's settings look like:

Matt Silverlock•11/25/24, 5:22 PM

This means that (docs: https://developers.cloudflare.com/queues/configuration/batching-retries/#batch-settings)

Both max_batch_size and max_batch_timeout work together. Whichever limit is reached first will trigger the delivery of a batch.

- The queue will wait for up to 1 message
- OR up to 5 seconds (which won't matter at a max_batch_size of 1)

It does not mean "one message every 5 seconds" - since you've set it to 1, it'll always deliver immediately if there's a message in the queue. If you need to rate limit on the consumer, consider the Rate Limit API: https://developers.cloudflare.com/workers/runtime-apis/bindings/rate-limit/

ajgeiss0702•11/25/24, 6:51 PM

is there any way to have concurrency increased?
I have no limit on concurrency, but concurrency almost always stays at 1. Went up to 2 for a few minutes, then back down to 1. It is going to take hours to get through the backlog with a single consumer

ajgeiss0702•11/25/24, 6:52 PM

im also running into a weird issue where batches are always taking 15 minutes, hitting the wall time limit (which in the dashboard shows up as a cpu limit error for some reason, even though cpu times stay around 300ms per batch), even though everything finishes way before then, which causes the backlog to take even longer to get through

queue id is

d1a94a7c4e4a4c23a6dbc74a8abbad11

d1a94a7c4e4a4c23a6dbc74a8abbad11

if it helps

ajgeiss0702•11/25/24, 6:53 PM

please ping me if you reply

Aajgeiss0702 im also running into a weird issue where batches are always taking 15 minutes, h...

ajgeiss0702•11/25/24, 10:03 PM

i suspect this has something to do with some request not being completed for some reason, which I was having issues with in the past. Is there a way to tell what it is waiting for?

oof2win2OP•11/26/24, 12:25 AM

is there a way to limit queue throughput? i have an upstream rate limit that i don't want to hit. i might have event spikes during the day and i don't want to drop requests so i want to limit myself to the set throughput (relatively speaking it's low, otherwise i wouldn't be asking for this)

oof2win2OP•11/26/24, 12:27 AM

like i want at most 10 messages per second. i have

max_concurrency

max_concurrency

on 1 but if the 10 messages in the batch complete faster than in 1 second then it doesn't work

Ooof2win2 like i want at most 10 messages per second. i have `max_concurrency` on 1 but if...

ajgeiss0702•11/26/24, 5:55 AM

you could simply wait a second after each message in the batch. Then if you have all 10 messages in a batch, then it shouldn't complete more than 1 per second

ajgeiss0702•11/26/24, 5:56 AM

for example with this function, just call

await wait(1000)

await wait(1000)

to wait a second

ajgeiss0702•11/26/24, 5:56 AM

this does not count as cpu time

ajgeiss0702•11/26/24, 5:57 AM

you could of course decrease that wait period if you want to process message faster than one per second (you could use 100ms to process at most 10 per second)

Aajgeiss0702 im also running into a weird issue where batches are always taking 15 minutes, h...

ac•11/26/24, 3:21 PM

I am seeing the same thing both with low concurrency and timeouts

ac•11/26/24, 3:22 PM

So the queue backlog is huge

Pranshu Maheshwari•11/27/24, 3:37 PM

There's a few reasons why consumers might not autoscale (documented here: https://developers.cloudflare.com/queues/configuration/consumer-concurrency/#why-are-my-consumers-not-autoscaling) (cc @ajgeiss0702)

If you're seeing timeouts @ac, it most likely means your Queue consumer is taking too long to process a message. You'll need to refactor your consumer to process the messages in under 30s of CPU time

Cloudflare Docs

Consumer concurrency | Cloudflare Queues

Consumer concurrency allows a consumer Worker processing messages from a queue to automatically scale out horizontally to keep up with the rate that messages are being written to a queue.

PPranshu Maheshwari There's a few reasons why consumers might not autoscale (documented here: https:...

ac•11/27/24, 3:39 PM

Thanks, I will take a look. Ultimately I believe the issue is browser rendering limits/the CF Browser Renderer holding onto processes for way longer than it's supposed to, which causes the whole queue to be backed up

PPranshu Maheshwari There's a few reasons why consumers might not autoscale (documented here: https:...

ajgeiss0702•11/27/24, 6:25 PM

I saw this, and I don't think it applies:
1. Max concurrency is not set
2. No errors are reported in the queues dashboard
3. batches are processed every 15 minutes (because of the other issue I mentioned) and it hasn't really scaled up

It has scaled up to 2 concurrency a few times, but hasn't gone past that, and it didnt stay on 2 for very long

xiechao•11/30/24, 6:55 AM

I would like to ask about billing related issues. I am currently facing a problem

xiechao•11/30/24, 6:56 AM

Remaining time on 11 × Images Stream Bundle Basic Stream storage per thousand minutes

xiechao•11/30/24, 6:56 AM

What is the fee?

1984 Ford Laser•11/30/24, 8:14 AM

?crossposting

SuperHelpflare•11/30/24, 8:14 AM

Please do not post your question in multiple channels/post it multiple times per the rules at #

welcome-and-rules. It creates confusion for people trying to help you and doesn't get your issue or question solved any faster.

Poyo•12/1/24, 1:10 PM

I'm trying to implement a simple queue. My worker sends a queue message to the consumer. I can see the message in the queue via the dashboard, but for some reason, the messages are just sitting in the queue, and processing doesn't start for 20 minutes. Do you have any ideas about what can cause this? I have logging and I can see that it is just sitting in the queue - no errors or anything.

Poyo•12/1/24, 1:12 PM

The timeouts and settings are default. I've spent a couple of hours but can't find a solution. I hope it is not an unannounced service disruption.

Poyo•12/1/24, 1:23 PM

Now it took a message to start processing - 4m 38s

export default { async queue(batch) { const ntfyUrl = 'https://ntfy.sh/doofy'; const message = batch.messages[0]; if (message) { try { const response = await fetch(ntfyUrl, { method: 'POST', body: message.body, }); if (response.ok) message.ack(); } catch (error) { console.error(`Failed to process message: ${error}`); } } }, };

is there a way with which i could limit the processing rate of a queue besides with concurrency + ba

Similar Threads

Similar Threads

Similar Threads