Does anyone have example code of how they handled multiple queues, single worker? Sounds like, bind

Ccontrol_crusher Does anyone have example code of how they handled multiple queues, single worker...

M

Producers are entirely separate from consumers.

- You can write to any queues you bind to as a producer.
- Your consumer will receive messages from only the queues it is configured as a consumer for.
- messages will be batched by queue, but each batch can be from any of the queues you have bound to.

M

Matt Silverlock•7/6/24, 1:05 PM

https://developers.cloudflare.com/queues/configuration/javascript-apis/#consumer and https://developers.cloudflare.com/queues/reference/how-queues-works/#create-a-consumer-worker

B

Ben•7/8/24, 5:01 PM

When queues go out of beta will they be available under workers free plan?

BBen When queues go out of beta will they be available under workers free plan?

P

Pranshu Maheshwari•7/8/24, 6:54 PM

we intend to make queues available under the free plan (with limits of course) in the future, although we don't have a timeline yet

PPranshu Maheshwari we intend to make queues available under the free plan (with limits of course) i...

B

Ben•7/8/24, 6:54 PM

sounds great!

B

BumblebeeSquare•7/8/24, 8:22 PM

is there message group like feature in queue ? i.e sqs message group

B

Brett Willis•7/9/24, 4:25 AM

It would be great to be able to manually pause/resume queues (collect from producers but pause dispatch to consumers) and also to purge queue messages. This help both in development and in production e.g. maintenance/migration, disaster recovery, or when dependent services are experiencing downtime. Queues are good for decoupling services for this very reason, and it's good to be able to intervene in emergencies!

M

minty•7/9/24, 8:30 PM

is it possible to consume messages from a queue from a non-worker piece of code? i.e. the way you can consume messages from SQS using their SDK in the language of your choice via polling/fetching?

Mminty is it possible to consume messages from a queue from a non-worker piece of code?...

H

Hello, I’m Allie!•7/9/24, 8:37 PM

Yes, see: https://developers.cloudflare.com/queues/configuration/pull-consumers

Cloudflare Docs

Cloudflare Queues - Pull consumers · Cloudflare Queues

A pull-based consumer allows you to pull from a queue over HTTP from any environment and/or programming language outside of Cloudflare Workers. A …

D

daguang•7/11/24, 10:53 AM

@Hard@Work @HardlyWorkin' Hello, when I use R2 GET data when consumers use R2 GET data, it will cause the queue to complete it directly. I tried to use Try Catch to capture the error, but I found nothing. Is this my configuration file wrong? or a bug ? (And this logic is normal in the normal worker fetch)

K

kian•7/11/24, 4:41 PM

you're returning errors rather than throwing them?

K

kian•7/11/24, 4:41 PM

:meowthinking:

Ccarlos-alberto Since ~18:00 yesterday one of our queues has been erroring with `Queue send fail...

A

ajgeiss0702•7/11/24, 11:38 PM

I am now also having this issue. Was there any workaround other than deploying a new queue and worker?

A

ajgeiss0702•7/11/24, 11:44 PM

trying to delete the worker throws "script still in use as a consumer for a queue"

A

ajgeiss0702•7/12/24, 1:37 AM

after manually removing the queue triggers, deleting the worker, and redeploying, the issue was fixed. Still annoying that was an issue though

3

33335•7/12/24, 2:38 AM

Kkian you're returning errors rather than throwing them?

D

daguang•7/12/24, 5:32 AM

I have try catch at the call, but I did not print the wrong log

Aajgeiss0702 I am now also having this issue. Was there any workaround other than deploying a...

P

Pranshu Maheshwari•7/12/24, 1:58 PM

hmm could you share your account ID & queue ID? Both are safe to share here.

A

ajgeiss0702•7/12/24, 1:58 PM

yes, 1m

A

ajgeiss0702•7/12/24, 1:59 PM

account id: f55b85c8a963663b11036975203c63c0

A

ajgeiss0702•7/12/24, 2:00 PM

and all of a sudden im not allowed to access the queues page in the dashboard?

A

ajgeiss0702•7/12/24, 2:00 PM

oh it just went through

A

ajgeiss0702•7/12/24, 2:00 PM

queue id: d1a94a7c4e4a4c23a6dbc74a8abbad11

WWallslide Is there a way to set a throughput limit. Like, don’t process more than 10 messa...

P

Pranshu Maheshwari•7/15/24, 2:25 PM

You could do it with pull based consumers: https://developers.cloudflare.com/queues/configuration/pull-consumers/

Cloudflare Docs

Cloudflare Queues - Pull consumers · Cloudflare Queues

A pull-based consumer allows you to pull from a queue over HTTP from any environment and/or programming language outside of Cloudflare Workers. A …

R

rcoopr•7/15/24, 9:27 PM

I think the answer to this is simply a no, but might as well ask. Is it possible with queues to process a task as soon as a single message is recieved, and dismiss the remaining messages that join the same batch?

I'm imagining the queue as a debounce of sorts, and this would be the

leading

leading

option for it.

Use case is something akin to logging, but latency is quite important here. Using batch timeout of ~3-5s if its relevant

Rrcoopr I think the answer to this is simply a no, but might as well ask. Is it possible...

M

Matt Silverlock•7/16/24, 4:57 AM

Why not set the max batch size = 1? Or just ack() all the messages you don’t care about in a batch (which means they will deleted) and only “process” what you want?

MMatt Silverlock Why not set the max batch size = 1? Or just ack() all the messages you don’t car...

R

rcoopr•7/16/24, 7:19 AM

I have a lot of duplicate API requests coming from separate instances of the same web page (going to some external API and don’t care about the response) and I’m looking to reduce the load on the external API and ‘dedupe’ requests via CDN/workers/queues. So a batch size of 1 would be like using the API directly I guess.

R

rcoopr•7/16/24, 7:20 AM

Basically I want:
Receive first api request: forward it on
Any duplicates within the next 3s, ignore

edit: ideally with POST requests, and the duplicate calls are triggered, in theory, at the same time. So when I tried simple caching with GET requests, the cache wasn't updated before duplicates came in

Hhannes I think a better approach is to just use in-memory cache of worker for dedupe-de...

R

rcoopr•7/16/24, 4:03 PM

is that just using the cache api? I tested using

caches.default

caches.default

but the requests seemd to hit before the cache write and subsequent read had updated

R

rcoopr•7/16/24, 4:04 PM

although that was in worst-case testing

R

rcoopr•7/16/24, 4:06 PM

well it would be a much better fit if it works as you say

A

Avat•7/17/24, 8:21 AM

Hi, I have a setup where single worker is both producer and consumer. I'm sending ~3k messages to queue, but only around 130 are being received.

No error is being returned in logs, how can I debug this?

Sharing my fetch function

fetch.js1.21KB

A

Avat•7/17/24, 8:22 AM

It seems like it's stuck awaiting for queue in

                        await env.QUEUE.send(message)

                        await env.QUEUE.send(message)

A

Avat•7/17/24, 8:38 AM

Does sending to queue counts as subrequest? So then I'm limited to 100 messages only?

AAvat It seems like it's stuck awaiting for queue in ` await en...

R

rcoopr•7/17/24, 9:05 AM

you can write tests for it https://developers.cloudflare.com/workers/testing/vitest-integration/test-apis/

Cloudflare Docs

Test APIs · Cloudflare Workers docs

Runtime helpers for writing tests, exported from the

cloudflare:test

cloudflare:test

module.

AAvat Hi, I have a setup where single worker is both producer and consumer. I'm sendin...

S

Sam•7/17/24, 9:08 AM

Also, why are you not sending the messages in a bulk?

SSam Also, why are you not sending the messages in a bulk?

A

Avat•7/17/24, 9:20 AM

Currently changing my code to send 90 messages at once to queue.
Consumer worker has to call API for every message, so I believe I'm limited by subrequests there, too?

A

anirudh•7/17/24, 12:23 PM

We are seeing around 700-800ms time to push to queue from the producer worker, is that normal? Seems a bit high...

D

dsp•7/17/24, 10:43 PM

Hello everyone. I'm using Queues to process WhatsApp Business API Webhooks, so, when I receive the batch, I can ack and keep going. For my surprise, I cannot receive any messages, until the entire batch is consummed. I thought max_concurrency could solve this, but as I see, It's not working. Here is my configuration:

[[queues.consumers]]
queue = "my-webhook"
max_batch_size = 100
max_batch_timeout = 3 #seconds
max_retries = 3
retry_delay = 30 #seconds
dead_letter_queue = "my-webhook-dlq"
max_concurrency = 20

[[queues.consumers]]
queue = "my-webhook"
max_batch_size = 100
max_batch_timeout = 3 #seconds
max_retries = 3
retry_delay = 30 #seconds
dead_letter_queue = "my-webhook-dlq"
max_concurrency = 20

[[queues.consumers]]
queue = "my-webhook"
max_batch_size = 100
max_batch_timeout = 3 #seconds
max_retries = 3
retry_delay = 30 #seconds
dead_letter_queue = "my-webhook-dlq"
max_concurrency = 20

[[queues.consumers]]
queue = "my-webhook"
max_batch_size = 100
max_batch_timeout = 3 #seconds
max_retries = 3
retry_delay = 30 #seconds
dead_letter_queue = "my-webhook-dlq"
max_concurrency = 20

So, say I reeive 10 messages in 3 seconds and start processing it ... Event if I use

ctx.waitUntil(promisse)

ctx.waitUntil(promisse)

followed by a

msg.ack{}

msg.ack{}

it does not receive other batches until all messages on this batch finishes processing.

My handler

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      ctx.waitUntil(processBody(message.body))
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      ctx.waitUntil(processBody(message.body))
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      ctx.waitUntil(processBody(message.body))
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      ctx.waitUntil(processBody(message.body))
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

Any tips on this?

Ddsp Hello everyone. I'm using Queues to process WhatsApp Business API Webhooks, so, ...

A

anirudh•7/18/24, 4:24 AM

You don't have to do the processing inside

ctx.waitUntil

ctx.waitUntil

Aanirudh You don't have to do the processing inside `ctx.waitUntil`

D

dsp•7/18/24, 11:46 AM

I changed it to

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      //Lots of seconds because call AI APIs and other stuff
      await processBody(message.body) 
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      //Lots of seconds because call AI APIs and other stuff
      await processBody(message.body) 
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      //Lots of seconds because call AI APIs and other stuff
      await processBody(message.body) 
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

async queue(batch, env, ctx) { 
  for (const message of batch.messages) {
    try {
      //Lots of seconds because call AI APIs and other stuff
      await processBody(message.body) 
      message.ack()
    } catch (e) {
      message.retry()
    }
  }
}

but still the same result. No concurrency. Only starts another batch when the current batch ends

Ddsp I changed it to ```js async queue(batch, env, ctx) { for (const message of ...

C

cmrpyro•7/18/24, 12:14 PM

hullo i just across this today, could either be a local thing or swapping to

promise.allSettled

promise.allSettled

, eg

async queue(batch, env, ctx) { 
  await Promise.allSettled(
    batch.messages.map((message) => processBody(message.body)
      .then(() => {
        // maybe `res?.ok` if its a fetch or something
        message.ack()
      })
      .catch(() => {
        message.retry()
      })
  )).finally(() => {
    console.log('batch completed')
  })
}

async queue(batch, env, ctx) { 
  await Promise.allSettled(
    batch.messages.map((message) => processBody(message.body)
      .then(() => {
        // maybe `res?.ok` if its a fetch or something
        message.ack()
      })
      .catch(() => {
        message.retry()
      })
  )).finally(() => {
    console.log('batch completed')
  })
}

async queue(batch, env, ctx) { 
  await Promise.allSettled(
    batch.messages.map((message) => processBody(message.body)
      .then(() => {
        // maybe `res?.ok` if its a fetch or something
        message.ack()
      })
      .catch(() => {
        message.retry()
      })
  )).finally(() => {
    console.log('batch completed')
  })
}

async queue(batch, env, ctx) { 
  await Promise.allSettled(
    batch.messages.map((message) => processBody(message.body)
      .then(() => {
        // maybe `res?.ok` if its a fetch or something
        message.ack()
      })
      .catch(() => {
        message.retry()
      })
  )).finally(() => {
    console.log('batch completed')
  })
}

Ccmrpyro hullo i just across this today, could either be a local thing or swapping to `pr...

D

dsp•7/18/24, 12:17 PM

Thanks. But how it will differ in the thing of no new batches are being triggered while I have some message still being processed?

Ddsp Thanks. But how it will differ in the thing of no new batches are being triggere...

C

cmrpyro•7/18/24, 12:21 PM

is this a local or a prod thing? docs make it sound like it will just scale and be concurrent on their platform but unsure locally.

assuming this doesn't get rate limited by a third party this batch will be done as soon as the slowest is done (parallel) rather than doing each message individually

Ccmrpyro is this a local or a prod thing? docs make it sound like it will just scale and ...

D

dsp•7/18/24, 12:21 PM

prod. Let me explain a litte bit more. I'm currently using Make (Integromat) to handle WhatsApp webhook messages. I changed to a Worker Endpoint, using queues, but stucked on processing

Ddsp prod. Let me explain a litte bit more. I'm currently using Make (Integromat) to ...

C

cmrpyro•7/18/24, 12:22 PM

ah dang. ok tapping out sadly

nah makes plenty of sense im wrestling with the same thing now

Does anyone have example code of how they handled multiple queues, single worker? Sounds like, bind

Similar Threads

Similar Threads

Similar Threads