my worker consumes from multiple queues and I want to switch based on the queue name, but I'd rather

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.queuebatch.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...

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

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

johtsoOP•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 startsWithstartsWith 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://...

johtsoOP•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 d1a94a7c4e4a4c23a6dbc74a8abbad11d1a94a7c4e4a4c23a6dbc74a8abbad11 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?

oof2win2•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)

oof2win2•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

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

Okay, I found a problem - the queue exceeded the CPU limit when the puppeteer was imported.

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

Found a way to fix it - don't use the browser inside queue but move it to a separate worker. Check the docs for Rendering Worker with Durable Object

ajgeiss0702•12/2/24, 12:45 AM

i wonder if thats the fix to my problem

ajgeiss0702•12/2/24, 3:57 AM

wow i think that might have solved my problem too

ajgeiss0702•12/2/24, 3:57 AM

so it seems like browser rendering has some issue with queues for some reason

ajgeiss0702•12/2/24, 3:58 AM

would have been nice to know this before black friday week, but better late than never

ajgeiss0702•12/2/24, 4:04 AM

its the first time ive seen this number in weeks! so beautiful!

PPoyo Found a way to fix it - don't use the browser inside queue but move it to a sepa...

ajgeiss0702•12/2/24, 4:11 AM

thank you very much. You just solved an issue i've been having for weeks and could not find a solution for!

NoGameNoFun•12/2/24, 6:54 AM

hi, i trying to set the cpu ms limit for the worker that run as consumer of a queueu, based on the documentation, it suppose can set a max limit of 15 minutes of cpu run limit, but when i try to set via wrangler or even using the dashboard, both i got is max 30,000 ms only

Aajgeiss0702 thank you very much. You just solved an issue i've been having for weeks and cou...

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

No problem. It has only taken me two days

I think the issue here is that the queue is not a standard worker. It has extra limitations that nobody is mentioning anywhere. Something to add to the docs as it will save so much time

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

By the way, when you run a browser in the queue directly, it keeps spinning until it exhausts all the CPU, and only after that does it time out. That is why we were seeing these random long waiting times. The browser blocks the consumer from execution until it times out, but it is my guess. I'm glad that I was able to help.

NNoGameNoFun hi, i trying to set the cpu ms limit for the worker that run as consumer of a qu...

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

My assumption is that a queue consumer is not a standard worker. I would also use a Durable Object - basically a separate normal worker connected to your queue consumer

queues

PPoyo By the way, when you run a browser in the queue directly, it keeps spinning unti...

ajgeiss0702•12/2/24, 4:34 PM

I can basically confirm it is some issue with browser rendering + queues. The only thing I changed was moving the browser rendering logic to a DO and calling that instead of it just being a function in my consumer. The rest of the logic is the same, and I have not had the issue since making that change

Sami•12/3/24, 3:07 AM

Hi, is there a way to find how many outstanding items/messages are present in a queue? Especially, to use with pull-workers.

juanferreras•12/3/24, 1:51 PM

I thought queues was not working reliably because I was seeing some unprocessed rows in my database from time to time.
Actually, I had a

try/catch

try/catch

on my consumer worker so I was swallowing the errors myself haha. Turns out I was the unreliable one and queues works great

go•12/4/24, 10:19 AM

I am using a cron once a day to fetch items from my database, add them to a queue, and process the items. However, as the cron is adding tens of thousands of items to the queue, I am reaching the limit of worker invocations I can do (1000 per worker). What are recommended ways to handle this kind of large batch processing in Workers?

my worker consumes from multiple queues and I want to switch based on the queue name, but I'd rather

Similar Threads

Similar Threads

Similar Threads