Is there a way to consume/product queues locally when using Vite as well? I've tried having `vite` a

Is there a way to consume/product queues locally when using Vite as well? I've tried having

vite

vite

and

wrangler dev

wrangler dev

running alongside each other but that doesn't seem to work

DDanGamble Is there a way to consume/product queues locally when using Vite as well? I've t...

Hello, I’m Allie!•11/4/24, 3:05 PM

You should be able to push into a Queue, assuming your framework supports it

Hello, I’m Allie!•11/4/24, 3:05 PM

But not consume

DDanGamble Is there a way to consume/product queues locally when using Vite as well? I've t...

James•11/4/24, 5:16 PM

when running locally i proxy the push to the queue worker's fetch which pushes to its local queue to then also consume

JJames when running locally i proxy the push to the queue worker's fetch which pushes t...

DanGambleOP•11/4/24, 7:45 PM

Do you have an example of this?

HHello, I’m Allie!You should be able to push into a Queue, assuming your framework supports it

DanGambleOP•11/4/24, 7:46 PM

Yea, I can push to it but I'm looking to consume it as well. At the moment I'm setting up a 2nd worker and using a service binding which just feels cumbersome

DDanGamble Do you have an example of this?

James•11/4/24, 7:48 PM

my main app is a pages project, and i have a separate worker for consuming the queue, on the qworker, i have a fetch which accepts the same payload as the queue send. if the origin worker is the same as the consumer worker, the queue works as expected locally, i only send to the fetch if my env is local, otherwise my pages app sends directly to the queue

DDanGamble Yea, I can push to it but I'm looking to consume it as well. At the moment I'm s...

Hello, I’m Allie!•11/4/24, 7:49 PM

Wait, so are you using a framework in Vite?

Hello, I’m Allie!•11/4/24, 7:49 PM

Or...

HHello, I’m Allie!Wait, so are you using a framework in Vite?

DanGambleOP•11/4/24, 7:50 PM

Yea, remix

DDanGamble Yea, remix

Hello, I’m Allie!•11/4/24, 7:50 PM

That is not currently possible, for dev anyway

Hello, I’m Allie!•11/4/24, 7:51 PM

The Frameworks team is working on a way to allow Vite-based dev servers to run in

workerd

workerd

, which would also allow them to expose non-HTTP event handlers(like Queues)

DanGambleOP•11/4/24, 7:51 PM

I thought as much. I imagine that's this: https://github.com/flarelabs-net/vite-plugin-cloudflare. If so, can't wait for it!

DoggeSlapper•11/7/24, 4:26 PM

Queue wrangler dev concurrency is 1 right?

Baltazar•11/7/24, 5:02 PM

does someone have a code example with pull consumers?

oof2win2•11/7/24, 7:21 PM

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

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

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

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

oof2win2•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 TimeExceeded 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 timeExceeded 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.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...

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

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 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://...

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

Is there a way to consume/product queues locally when using Vite as well? I've tried having `vite` a

Similar Threads

Similar Threads

Similar Threads