the limits are documented here: https://developers.cloudflare.com/queues/platform/limits/

KKam I'm seeing queue `send` and `sendBatch` calls taking anywhere from ~250ms to ove...

B

Brett Willis•8/12/24, 8:47 PM

Can't comment on those latencies myself, but you could consider using

ctx.waitUntil()

ctx.waitUntil()

to more cheaply and easily defer those operations asynchronously without holding back the response.

JJohn Spurlock the limits are documented here: https://developers.cloudflare.com/queues/platfor...

F

flyingshoes•8/12/24, 8:58 PM

got it -- so 400 messages per second. thank you.

KKam I'm seeing queue `send` and `sendBatch` calls taking anywhere from ~250ms to ove...

P

Pranshu Maheshwari•8/12/24, 10:13 PM

it's a bit higher than expected, but not out of the ordinary. We're working on improvements that should improve latency for sending messages to a queue. Those improvements should ship by Oct.

KKam Thanks for the info Pranshu! Any idea what latency you are targeting with the im...

P

Pranshu Maheshwari•8/13/24, 3:11 PM

It's too soon to share specific targets, but stay tuned!

N

NoGameNoFun•8/14/24, 3:26 AM

Hi everyone, I like the worker and queue features in cloudflare, and been trying out about few weeks already.

I got a question, I noticed the consumer for the queue, after some time (like 2 days), it will stop processing the queue message. (the backlog size increase and doesnt get consume)

I attached the screenshot on the consumer setting, is this a bug or is just because of my consumer setting?

J

jordanfinners•8/14/24, 9:13 PM

Please can I provide some feedback/feature requests for cloudflare queues

Firstly the dev experience inside workers is pretty great! It would be awesome if it could work with Pages functions so I don't have to deploy separate consumers but thats a nit really.

The metrics in the dashboard are useful but its missing the main one for me, how many messages are in the queue at that point in time, especially important for deadletters
Redriving of deadletter messages, either a button in the UI or a way of doing it via HTTP or JS.
Currently you can pull the messages from the queue via HTTP but can't acknowledge them or push back into the queue locally, meaning I would need to deploy a worker just to redrive each queue which seems excessive and I need to control what I'm redriving or not.
I had to do 200 messages manually this morning
Alerts on the queue length, ie. If messages drop into a deadletter
Would be helpful if the error message was attached to the queue message in some way like SQS does
Being able to send multiple messages via the UI either file upload or pasting in
Documentation is missing for the Preview endpoint in the Queue REST API docs
Its only happened once but I've had one message not get sent, even though it said it has, which worries me slightly.
If possible increase the batch size, up to 200 would be helpful
Alongside this if possible to increase the timeout period as well, I'm running async reports and these can take several minutes to process and so they are timing out/hitting worker limits.

Happy to talk this through or provide more info if

Wwindygod9986 Does anyone know if cloudflare has any plans to support queue consumers on page...

M

Matt Silverlock•8/17/24, 12:38 AM

Not in the near term. Pages doesn’t have the concept of any non-request handler (cron, queues, tail)

L

lulupineapple•8/18/24, 2:06 PM

Hi, how do I deal with queue consumer timeout issue? i.e. make sure queue consumer completed the task with payload? Is there anyway to have queue retry calling consumer unless consumer explicitly acknowledge the payload was processed susessfully?

O

oof2win2•8/21/24, 12:21 PM

how can i set proper types for a queue?

O

oof2win2•8/21/24, 12:23 PM

like it seems that i need to change it on the generated env types but then i can't have them be generated

O

oof2win2•8/21/24, 12:45 PM

more importantly it doesn't work for receiving the messages via the same queue

Ooof2win2 more importantly it doesn't work for receiving the messages via the same queue

M

Matt Silverlock•8/21/24, 10:41 PM

The TypeScript types for the

queue

queue

handler (your consumer) accept a type parameter - so

async queue<T>(batch: MessageBatch, env: Env)

async queue<T>(batch: MessageBatch, env: Env)

- which will mean that each message in

MessageBatch

MessageBatch

has a typed payload based on your type parameter

O

oof2win2•8/22/24, 5:55 AM

thanks!

S

snapo•8/22/24, 7:30 AM

Hey people!

When developing locally, I have configured my queue with following retry timeouts and counts:

max_retries = 10,  
retry_delay = 30

max_retries = 10,  
retry_delay = 30

I noticed that for some unknown reason, the retry logic only attempts up to 3 times. If I lower the retry_delay to 10 seconds, I can reach 6th attempt. If I finally lower the delay to 2 seconds, I can reach 10 retry attempts which is expected.

Any ideas, why the 30s delay is not working and 2 seconds is?

K

Kaique Nunes•8/22/24, 10:31 AM

it's possible to list messages inside dead queue?

KKaique Nunes it's possible to list messages inside dead queue?

S

snapo•8/22/24, 11:34 AM

You can do it via the dashboard. If you want to use http then you could configure a pull consumer for the dlq

Ssnapo You can do it via the dashboard. If you want to use http then you could configur...

K

Kaique Nunes•8/22/24, 11:59 AM

So I would need to implement my own api to retrieve the dlq messages?

S

snapo•8/22/24, 12:00 PM

No no. See here: https://developers.cloudflare.com/queues/configuration/pull-consumers/

Cloudflare Docs

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 pull-based consumer can be useful when your message consumption rate is limited by upstream infrastructure or long-running tasks.

S

snapo•8/22/24, 12:01 PM

https://developers.cloudflare.com/api/operations/queue-v2-messages-pull

Cloudflare API Documentation

Interact with Cloudflare's products and services via the Cloudflare API

K

Kaique Nunes•8/22/24, 12:08 PM

Thanks

KKaique Nunes it's possible to list messages inside dead queue?

P

Pranshu Maheshwari•8/23/24, 2:35 PM

Dead letter queues behave just like normal queues. You could setup a consumer, just like you would with a normal queue. It doesn't necessarily need to be a pull consumer; you could use a worker too.

PPranshu Maheshwari Dead letter queues behave just like normal queues. You could setup a consumer, j...

K

Kaique Nunes•8/23/24, 2:46 PM

But I would need to store the messages somewhere

K

Kaique Nunes•8/23/24, 2:47 PM

I don’t want to need to add an extra db only for this

K

Kaique Nunes•8/23/24, 2:47 PM

Ideally queues could allows us to list pending messages

P

piffie•8/25/24, 9:56 PM

is there a way of getting the name of the queue, when you have the queue object?

C

clearwater4227•8/25/24, 9:57 PM

Hi everyone! I am starting looking into using queues but have one question...can I configure a worker script as a queue consumer through terraform? I havent found the setup yet..still looking but if anyone here knows or can point me to the info...

C

clearwater4227•8/25/24, 9:58 PM

it looks like I can configure the queue reference on a worker to send a message but I need to configure the consumer through terraform also.

C

clearwater4227•8/25/24, 10:04 PM

or is it just a worker with the queue() function? hmm...Ill try it

C

clearwater4227•8/25/24, 10:21 PM

hmmm...its not picking up my worker as a consumer... is there a way to configure a queue consumer through terraform?

P

piffie•8/25/24, 10:51 PM

never used terraform for it - sorry

B

berkinovish•8/26/24, 8:42 AM

Is there a way to force a queue to increase concurrency, or to set a minimum concurrency?

I'm seeing the following behaviour now:

I have a queue where only once every 1-5 minutes, a message is queued. A message takes about 30 seconds wall time to be processed. Maximum consumer concurrency is on default.

But, when a message comes in that all of a sudden takes 30 minutes to process, the whole queue will be stuck until that one message is processed (that's the behaviour I'm observing at least). This behaviour is in line with the docs, since Queues will only check after processing a batch of messages if the concurrent consumers should be adjusted.

Very annoying behaviour of course - because of a slow message, I suddenly have messages waiting for 30 minutes in the queue. Only after the message is processed, the concurrency will be updated and all waiting messages will be handled. Being able to set a minimum concurrency would be a big help here

J

John SpurlockOP•8/26/24, 4:17 PM

anyone else see queue send failures across of dozens of colos at around 2024-08-26 06:21:46 UTC?

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error

Haven't seen anything across so many colos at once like this before

JJohn Spurlock anyone else see queue send failures across of dozens of colos at around 2024-08-...

M

Matt Silverlock•8/26/24, 4:42 PM

@Pranshu Maheshwari

JJohn Spurlock anyone else see queue send failures across of dozens of colos at around 2024-08-...

M

Matt Silverlock•8/26/24, 4:45 PM

FWIW I don't see anything in top-level metrics for Queues — no meaningful dip in write or read throughput around 6:21 UTC or increase in 5xx error ratio. Pranshu can dig in with the team.

MMatt Silverlock FWIW I don't see anything in top-level metrics for Queues — no meaningful dip in...

J

John SpurlockOP•8/26/24, 4:50 PM

ok thanks, it was a relatively short duration (ended

06:22:09

06:22:09

06:22:09

06:22:09, but seems like essentially all sends failed (these are after client-side retries) during that time - worker connection to the rest of cf was ok (I was able to fallback to saving msgs to DO for later playback)

E

esselungo•8/27/24, 12:06 PM

i take the liberty to link that message here since it might be related to queues workers-and-pages-discussions

J

Julio Ribeiro•8/27/24, 1:40 PM

Hey team, whats up?

I have a use case for queues and want to know if my logic is correct:
- I have a producer that posts around 500 messages at once on the queue, using batches of 25 messages
- All batches are delayed incrementally until the last batch on increments of 30 seconds each batch (first batch for 0 seconds, the second for 30 seconds, the third for 60 seconds...)
I was doing this process previously with a D1 column for the queue and a scheduled workers that runs each minute, but wanted to try it on using queues
The consumer should receive the batches following the delays, right?
The problem I'm facing: All the batches get processed at once, not obeying the delay sent while creating the batches.
I need this delay on the batches due to Slack API limitation that I use to process these messages, avoiding receiving 429s.
Is there any problem on the CF side causing this problem, or the queue will process all messages at once even with the delays?

Here are some snippets so you can take a look at my code:

queue definition

queue definition

queue definition

queue definition:

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

producer

producer

:

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

consumer

consumer

:

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

C

Conner•8/27/24, 5:01 PM

I think I might be seeing some messages dropped from my queue? Some final data in our db is missing and I traced it back to this.

Cloudflare dashboard shows 321 messages successfully delivered
Logpush logs only show 313 events from that queue

All of these messages were sent in the same few seconds/min. What happened to the other 8 messages?

.

.gwapes•8/28/24, 11:23 AM

Hi, new to queues. I'm doing this:

await env.queue.send(
                data,
                { delaySeconds: 42300, contentType: "json" }
            ).then(() => console.log("done"));

await env.queue.send(
                data,
                { delaySeconds: 42300, contentType: "json" }
            ).then(() => console.log("done"));

await env.queue.send(
                data,
                { delaySeconds: 42300, contentType: "json" }
            ).then(() => console.log("done"));

await env.queue.send(
                data,
                { delaySeconds: 42300, contentType: "json" }
            ).then(() => console.log("done"));

However, whenever I view on the dashboard and list messages, nothing shows, however "done" shows in the live logs. Am I doing something wrong? Smaller times work though.

J

johtso•8/29/24, 6:24 PM

Shouldn't this page document the

attempts

attempts

attempts

attempts property on a

Message

Message

Message

Message? https://developers.cloudflare.com/queues/configuration/javascript-apis/

J

johtso•8/29/24, 6:25 PM

wanted to quickly check if the value is 0 or 1 the first time a message is received, but saw it wasn't documented at all

J

johtso•8/29/24, 6:38 PM

https://github.com/cloudflare/cloudflare-docs/issues/16542

GitHub

Queues: Message.attempts is undocumented · Issue #16542 · cloudflar...

Proposed changes Document the Messages.attempts property here: https://developers.cloudflare.com/queues/configuration/javascript-apis/ Subject Matter Message.attempts property of a queue message Co...

C

cloudkite•9/2/24, 12:08 AM

When you explicitly call message.retry() does this still respect the

max_retries

max_retries

max_retries

max_retries configuration setting in wrangler.toml?

C

cloudkite•9/2/24, 1:01 AM

It does't seem like the messages get redelivered when calling message.retry() with local wrangler dev?

L

lusine•9/3/24, 9:36 AM

Hi team, I need to use event notifications with R2 and queues when new objects are created. I have a job written in C#/.NET which is supposed to read messages from this queue. My question is: do you have a .NET SDK that I can use for receiving/acknowledging messages from queues? For example, I think R2 is compatible with AWS S3 SDK and I can use AmazonS3Client with R2, is that the case with queues? Can I use AmazonSqsClient with Cloudflare queues? Or what other options do I have?

the limits are documented here: https://developers.cloudflare.com/queues/platform/limits/

Similar Threads