FWIW I don't see anything in top-level metrics for Queues — no meaningful dip in write or read throu

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

J

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

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

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?

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

P

Pranshu Maheshwari•9/3/24, 2:34 PM

yes!

Llusine Hi team, I need to use event notifications with R2 and queues when new objects a...

M

Matt SilverlockOP•9/3/24, 11:33 PM

You can pull via HTTP - we don’t have a .NET SDK and the SQS API is far less of a defacto standard vs S3, and so we have no plans to support it. 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.

Ssouvikinator Hey guys I'm new to the Cloudflare ecosystem. Need some guidance. Here is my req...

H

Hard@Work•9/4/24, 9:17 AM

?crossposting

F

Flare•9/4/24, 9:17 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.

Ssouvikinator Hey guys I'm new to the Cloudflare ecosystem. Need some guidance. Here is my req...

H

harshil1712•9/4/24, 10:00 AM

I don't see why Queues won't be a good fit!

Ssouvikinator My concerns are batch processing, persistance and stability. I saw few posts abo...

H

harshil1712•9/4/24, 10:07 AM

I have built some demos with queues and have not faced the issues. One thing I have done to make sure that there is consistency is adding retries at the relevant places. You can also setup a dead-letter queue. Here's the doc that might be useful: https://developers.cloudflare.com/queues/configuration/batching-retries/#explicit-acknowledgement-and-retries

Cloudflare Docs

Batching, Retries and Delays | Cloudflare Queues

When configuring a consumer Worker for a queue, you can also define how messages are batched as they are delivered.

Ssouvikinator great I need to try it out. So I assume it's ready for production right?

H

harshil1712•9/4/24, 10:12 AM

It's going to be GA soon

Ssouvikinator Hey also if you could explain me a bit because I'm still figuring out how the wo...

H

harshil1712•9/4/24, 10:53 AM

You can have the producer and the consumer in the same worker or in separate ones. It really depends on how you want to architect your application. However, for now, when building locally, we recommend having them in the same worker.
https://developers.cloudflare.com/queues/configuration/local-development/#known-issues

The pricing docs have the answer on the execution: https://developers.cloudflare.com/queues/platform/pricing/

Cloudflare Docs

Pricing | Cloudflare Queues

Cloudflare Queues charges for the total number of operations against each of your queues during a given month.

Cloudflare Docs

Local Development | Cloudflare Queues

Queues support local development workflows using Wrangler, the command-line interface for Workers. Wrangler runs the same version of Queues as Cloudflare runs globally.

C

celer•9/4/24, 8:46 PM

Hi! Has anyone seen CPU Execution timeout on queue consumer event? In the worker limits docs they say the limit is 15 minutes of CPU time. But I always get timeout after 32,000ms (according to the worker cpu time dashboard). I could work around that, but I'd like to know if the docs are wrong or if it's a bug on my or CF side.

M

Michael Wittig•9/5/24, 12:49 PM

Hi! I deleted a bunch of R2 bucket notifications & buckets but I can not delete the queue that was referenced. I always receive the error "cannot delete queue bucketav-rhyglnin that serves as a target for event notifications". I have no buckets left so this is likely not the case. Are there any reqasons for that? Is there something happening in the background before I can delete the queue? If I look at the queue in the dashboard it says "This queue does not have a consumer". Clicking the delete button in the UI is broken as well. I can see the error in the browser logs but not the UI.

J

johtso•9/5/24, 2:42 PM

hmm, it seems like my messages are being retried immediately after an implicit retry instead of waiting the configured 60 seconds..

[[env.staging.queues.consumers]]
queue = "emails-staging"
max_batch_size = 1
max_concurrency = 1
dead_letter_queue = "emails-dlq-staging"
max_retries = 5
retry_delay = 60

[[env.staging.queues.consumers]]
queue = "emails-staging"
max_batch_size = 1
max_concurrency = 1
dead_letter_queue = "emails-dlq-staging"
max_retries = 5
retry_delay = 60

[[env.staging.queues.consumers]]
queue = "emails-staging"
max_batch_size = 1
max_concurrency = 1
dead_letter_queue = "emails-dlq-staging"
max_retries = 5
retry_delay = 60

[[env.staging.queues.consumers]]
queue = "emails-staging"
max_batch_size = 1
max_concurrency = 1
dead_letter_queue = "emails-dlq-staging"
max_retries = 5
retry_delay = 60

S

Stockton•9/7/24, 12:14 AM

What would be the benefits of having a producer and consumer in two different workers? Or vice versa?

M

micooz•9/8/24, 5:22 PM

Queue "consumer concurrency" feature does not seem to be suitable for scenarios that rely on sequential execution.

Y

ytsruh•9/10/24, 1:59 PM

Hey all, does anybody if there is a way to get both Queues & the Browser Rendering API working nicely together in local dev? Seems Queues doesn't support --remote but Browser Rendering only works when using --remote?

Yytsruh Hey all, does anybody if there is a way to get both Queues & the Browser Renderi...

S

Sam•9/10/24, 2:56 PM

Did not have your problem, but you could try to just put your queue script inside the fetch() one to test it out remotely.

SSam Did not have your problem, but you could try to just put your queue script insid...

Y

ytsruh•9/10/24, 3:02 PM

Thanks. I think we’re switching to scheduled instead. Our use case didn’t have massive throughput. Cheers for quick reply

OOle Melhus Hey! I'm seeing delays for up to 5-7 seconds between adding something to a queue...

H

harshil1712•9/11/24, 12:05 PM

Can you expan a bit more on this? Is the consumer queue not picking it up? Can you also share your wrangler.toml config for the queues?

P

psteinroe•9/12/24, 2:22 PM

hey! is there any (inofficial) goal for queues performance? will it ever be comparable to eg sqs? sending to the queue is currently multiiple 100ms, and it takes multiple seconds for a worker to receive it. and throughput is much much lower at the moment.

K

kcbrewron•9/13/24, 11:23 PM

I haven't run into messages dropping or batches dropping. I have built some interesting architecture where I have a message-producer API which abstracts all the applications from needing to know about how to send messages. We used a KV bound to the message producer API to know what queue to route to and bind the worker to any queue. I know there are limits to the binding, but this API has made it easy to onboard queues for a few different use cases. I'm ready for this to move to GA. What issues is the team waiting on solving before moving it to GA?

O

oof2win2•9/16/24, 9:40 AM

is there a plan to have messages be "grouped" and sent that way? i have a merchantId prop on my messages and i want to handle up to 20 messages per merchant, but there isn't really a way to do that right now

O

oof2win2•9/16/24, 9:41 AM

it would be really neat if i could do that honestly

V

VoidRanger•9/16/24, 10:15 PM

Hi there, just wanted to know if there is any way to see which buckets are sending notifications to a queue

T

tadhglewis•9/17/24, 1:07 PM

Couple queries:

Any plans to add a

Message.deadletter

Message.deadletter

Message.deadletter

Message.deadletter method? I have some use cases where I know the message will always fail and it needs manual triage.

It'd be great to have redrive functionality to move messages from the deadletters to processing queue once any issues with the deadletters have been resolved (e.g. temp outage). AWS SQS has similar functionality https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-dead-letter-queue-redrive.html

Learn how to configure a dead-letter queue redrive in Amazon SQS - ...

Learn how to configure an Amazon SQS dead-letter queue redrive to move messages from a dead-letter queue to a source queue or other standard queue.

T

tadhglewis•9/17/24, 1:10 PM

Also, are there plans to suppport passing a

Message

Message

Message

Message into an RPC Worker?

J

Jason Barry•9/18/24, 11:25 PM

I don't see wrangler.toml's equivalent of

[[queues.consumers]]

[[queues.consumers]]

[[queues.consumers]]

[[queues.consumers]] referenced in https://developers.cloudflare.com/workers/configuration/multipart-upload-metadata/#bindings -- is that intentional? if so, how does one configure a Worker uploaded from the API to be a queue consumer? I've added

"type": "queue"

"type": "queue"

"type": "queue"

"type": "queue" in my bindings metadata, but that only seems to register the worker as a queue producer

JJason Barry I don't see wrangler.toml's equivalent of `[[queues.consumers]]` referenced in h...

H

Hello, I’m Allie!•9/19/24, 6:33 AM

It's done via this API: https://developers.cloudflare.com/api/operations/queue-v2-create-queue-consumer

Cloudflare API Documentation

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

FWIW I don't see anything in top-level metrics for Queues — no meaningful dip in write or read throu

Similar Threads

Similar Threads

Similar Threads