Why not set the max batch size = 1? Or just ack() all the messages you don’t care about in a batch (

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

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)

                        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

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)

ctx.waitUntil(promisse)

ctx.waitUntil(promisse) followed by a

msg.ack{}

msg.ack{}

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

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

Ccmrpyro ah dang. ok tapping out sadly 😞 nah makes plenty of sense im wrestling with the...

D

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

Each batch is processed only in sequence.

I thought the concurrency feature (max limit 20 as I learned) could be used to scale lots of batches

Have to wait for the entire batch to process does not make any sense to me. It should have real concurrency batch process or I'm missing something.

I'm trying to follow the DOCs instructions here - https://developers.cloudflare.com/queues/configuration/consumer-concurrency/

Ddsp Each batch is processed only in sequence. I thought the concurrency feature (ma...

C

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

have you tried setting the batch size to something really small, forcing the backlog to grow? maybe its below the limit of making another concurrent worker. I too thought a batch would be in parallel

Ccmrpyro have you tried setting the batch size to something really small, forcing the bac...

D

dsp•7/18/24, 1:27 PM

For sure. If I set 1 and timeout to 0, only process 1 message at a time, and it's almost impossible to do anything useful

I'll probably create 10+ queues and random between then.

J

jordanfinners•7/18/24, 1:51 PM

Hey, would anyone at Cloudflare be able to help me please

I've got some messages that are missing

D

dsp•7/18/24, 7:14 PM

Can I have more than 1 worker consumming from the same Queue?

Ddsp Can I have more than 1 worker consumming from the same Queue?

R

rcoopr•7/18/24, 9:33 PM

how would it choose which worker to use? what's the purpose?

Rrcoopr how would it choose which worker to use? what's the purpose?

D

dsp•7/19/24, 2:21 AM

Concurrency. Can't make it work as the docs explain. My curretlly workaround is to have 10 queues with producers and consummers

Ddsp Concurrency. Can't make it work as the docs explain. My curretlly workaround is ...

P

Pranshu Maheshwari•7/19/24, 2:56 PM

What's the issue you're having right now?

You can setup pull consumers (https://developers.cloudflare.com/queues/configuration/pull-consumers/) if you want to manually control the level of concurrency. But the default push based consumers should automatically scale up your consumer workers based on demand.

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 …

PPranshu Maheshwari What's the issue you're having right now? You can setup pull consumers (https:...

D

dsp•7/19/24, 8:34 PM

Hi @Pranshu Maheshwari. How are you doing?

The push based should, but it doesn't

My currently workaround is to have 10 producers and consumers and distribute between then

Here is the config:

[[queues.producers]]
binding = "WEBHOOK_QUEUE_1"
queue = "webhook-1"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_2"
queue = "webhook-2"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_3"
queue = "webhook-N"

[[queues.consumers]]
queue = "webhook-1"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-2"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-N"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_1"
queue = "webhook-1"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_2"
queue = "webhook-2"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_3"
queue = "webhook-N"

[[queues.consumers]]
queue = "webhook-1"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-2"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-N"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_1"
queue = "webhook-1"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_2"
queue = "webhook-2"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_3"
queue = "webhook-N"

[[queues.consumers]]
queue = "webhook-1"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-2"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-N"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_1"
queue = "webhook-1"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_2"
queue = "webhook-2"

[[queues.producers]]
binding = "WEBHOOK_QUEUE_3"
queue = "webhook-N"

[[queues.consumers]]
queue = "webhook-1"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-2"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

[[queues.consumers]]
queue = "webhook-N"
max_batch_size = 10
max_batch_timeout = 2 
max_retries = 3
retry_delay = 30 
dead_letter_queue = "webhook-dlq"

Where N is just the following configs for both producers and consummers until 10.

If you have any tips ...

M

marko.333•7/20/24, 7:42 PM

Is there any plan to enhance the queue api to query the number of messages in the queue, or at least if it's empty or not? thx.

M

marko.333•7/21/24, 12:58 PM

There was a bit of a thread yesterday that discussed ongoing work to support dynamic provisioning/binding of D1 db's.
I was wondering if something similar is being worked on for Queue's (i'm hopeful

)?
d1-database

Mmarko.333 Is there any plan to enhance the queue api to query the number of messages in th...

J

John Spurlock•7/21/24, 3:01 PM

you can query the avg message size and/or avg number of messages in the backlog over time for a given queue using the graphql api (

queueBacklogAdaptiveGroups

queueBacklogAdaptiveGroups

queueBacklogAdaptiveGroups

queueBacklogAdaptiveGroups)

https://developers.cloudflare.com/analytics/graphql-api/

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

S

Shivek•7/21/24, 4:41 PM

You can set the batch size to 10 in wrangler toml?

Ddsp Each batch is processed only in sequence. I thought the concurrency feature (ma...

S

Shivek•7/21/24, 4:43 PM

Did you confirm this ? Are queue batches processes sequentially? I was assuming they would be processed parallely.

Ie, if the batch size is 100, and I have 1000 items, 10 workers will spawn and process all the items in one go

Ddsp Hi @Pranshu Maheshwari. How are you doing? The push based should, but it doesn...

S

Shivek•7/21/24, 4:46 PM

How did you verify that your consumers are not scaling?

B

berkinovish•7/22/24, 9:45 AM

I have a queue with concurrency = 1, because we're rate limited by an external API. I'd like to set up a system, where as soon as the call to the external API is completed, a new message gets pulled from the queue, while the previous worker is still handling persisting the data to the database.

Right now, if you

ack()

ack()

ack()

ack() a message, while you're still handling some stuff inside a

ctx.awaitUntil()

ctx.awaitUntil()

ctx.awaitUntil()

ctx.awaitUntil(), the queue will wait with handling the next batch until the

waitUntil

waitUntil

is fully handled.

Any idea on how we could solve this with queues?

Bberkinovish I have a queue with concurrency = 1, because we're rate limited by an external A...

H

Hard@Work•7/22/24, 9:48 AM

Have two queues, one that handles API calls, then another that handles responses?

HHard@Work Have two queues, one that handles API calls, then another that handles responses...

B

berkinovish•7/22/24, 9:52 AM

This works indeed, but pushing to the queue gets also awaited. Pushing a message on the queue can take up to 1 second for us

Bberkinovish This works indeed, but pushing to the queue gets also awaited. Pushing a message...

H

Hard@Work•7/22/24, 9:53 AM

Maybe set the batch sizes to be larger, so you can run multiple one after another?

H

Hard@Work•7/22/24, 9:53 AM

Not a perfect fix, but should help a bit

HHard@Work Not a perfect fix, but should help a bit

B

berkinovish•7/22/24, 9:55 AM

Yeah, would help a bit, was wondering if we could get even more out of it. Batching + pushing to a separate queue is the best we can do then probably

Bberkinovish I have a queue with concurrency = 1, because we're rate limited by an external A...

D

dsp•7/22/24, 12:11 PM

That's exactally my problem. concurrency is a Myth. you have to wait for the whole batch to finish.

Ddsp That's exactally my problem. concurrency is a Myth. you have to wait for the who...

M

Matt SilverlockOP•7/22/24, 10:28 PM

That’s not true: we don’t (currently) autoscale if you have a consumer that is especially slow - the “wall clock” time of the consumer doesn’t yet feed into our autoscaling alg.

Batches are consumed concurrently.

You could consider working around this in the meantime by setting concurrency = 20 (the current maximum).

MMatt Silverlock That’s not true: we don’t (currently) _autoscale_ if you have a consumer that is...

D

dsp•7/23/24, 12:24 AM

Already tried. Doesn't work

Ddsp Already tried. Doesn't work 😦

M

Matt SilverlockOP•7/23/24, 12:25 AM

Have you opened a ticket so we can take a deeper look? (can’t see if you have shared your account & queue IDs here)

Why not set the max batch size = 1? Or just ack() all the messages you don’t care about in a batch (

Similar Threads

Similar Threads

Similar Threads