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

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

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

HHello, I’m Allie!It's done via this API: https://developers.cloudflare.com/api/operations/queue-v...

J

Jason Barry•9/19/24, 2:42 PM

ah, I forgot to mention that I was trying to create queue consumers for user-uploaded scripts. I filed https://github.com/cloudflare/workers-sdk/issues/6758

JJason Barry ah, I forgot to mention that I was trying to create queue consumers for user-upl...

H

Hello, I’m Allie!•9/19/24, 2:44 PM

Via WfP?

HHello, I’m Allie!Via WfP?

J

Jason Barry•9/19/24, 2:44 PM

yep

JJason Barry yep

H

Hard@Work•9/19/24, 3:22 PM

Can you not just have the dispatcher call the user Worker?

H

Hard@Work•9/19/24, 3:23 PM

Also, I don't think WfP actually supports any kind of events other than one's sent by the dispatcher atm

HHard@Work Can you not just have the dispatcher call the user Worker?

J

Jason Barry•9/19/24, 3:32 PM

that’s the direction I was heading, by passing messages received in the dispatcher’s queue consumer to the user worker via a POST fetch. but how can the user worker call message.retry(), since the messages will be serialized into JSON?

JJason Barry that’s the direction I was heading, by passing messages received in the dispatch...

H

Hard@Work•9/19/24, 3:35 PM

It returns an object back with which messages to retry/ack

A

ajgeiss0702•9/20/24, 2:09 AM

it is just me, or does

delaySeconds

delaySeconds

delaySeconds

delaySeconds no longer work?

A

ajgeiss0702•9/20/24, 2:10 AM

it looks like it stopped working on 9/17 around 18:00 UTC

A

ajgeiss0702•9/20/24, 2:25 AM

its still listed on the dev docs still, so it wasnt removed, was it?

A

ajgeiss0702•9/20/24, 7:02 PM

please ping me if you reply

A

ajgeiss0702•9/20/24, 7:03 PM

im currently getting ratelimited by an external service because things that are supposed to be delayed (using queues) are all getting run at once

S

Sam•9/21/24, 2:52 AM

Hey the time between my message getting sent to queue -> being processed by my working is taking a few seconds before it even starts being processed. Anyway to speed this up?

S

Sam•9/21/24, 2:57 AM

EEmilio The default is 5000ms and yes you can set it to 0

S

Sam•9/21/24, 3:23 AM

Seems 1000/1s is the minimum value in dashboard UI and cant be set via wrangler.toml

U

UNOwen•9/22/24, 9:49 AM

Hey folks, fresh Queues user here. I am having issues with queue messages not being processed after they are sent back for a

.retry(12 hours)

.retry(12 hours)

.retry(12 hours)

.retry(12 hours), and I can't figure out whether I've mis-configured something or am hitting an edge case.

I produce queue messages in batch from my Pages website, it is basically a set of payloads for emails to be sent out, some to be sent out immediately (message has a field

after: now.toISO()

after: now.toISO()

after: now.toISO()

after: now.toISO()), and some upon delay (message has a field

after: remindOn.toISO()

after: remindOn.toISO()

after: remindOn.toISO()

after: remindOn.toISO(), typically 3-5 days from now). The are 0-3 batches of these messages per day sent out worldwide, as in these are not high volume.

The messages are received by a Workers consumer, generated from a standard Cloudflare workers template. The relevant bit is the

.retry({delaySeconds: X})

.retry({delaySeconds: X})

.retry({delaySeconds: X})

.retry({delaySeconds: X}) of messages which should not be processed yet: if the

after

after

field of the payload message is in the future, I am sending the message back for a retry of up to 12 hours, because it is the documented maximum I found.

What happens, is that the messages I've sent for a long retry don't seem to process until there is a fresh new unrelated message added to the queue, which seems to trigger the re-processing of

.retry()

.retry()

.retry()

.retry()-ed messages. This doesn't seem to happen globally either, but rather per region, as in that "fresh new unrelated message" will only trigger the re-processing of geographically proximate messages. E.g. I had somebody in Germany visit my website on Sept 11 and fill out the booking form for Sept 16 lesson; they received the "immediate" messages, but the "reminders" got stuck in limbo until Sept 11 until a completely different person somewhere in Russian filled out the form which triggered new messages to be added to the queue.

UUNOwen Hey folks, fresh Queues user here. I am having issues with queue messages not be...

U

UNOwen•9/22/24, 9:49 AM

Message sending batch:

    await platform.env.bookings.sendBatch([
        { body: { action: 'notify', target: 'student',  via: 'email', after: now.toISO(),      message: 'requested_lesson', ...booking }},
        ...2 messages omitted...
      message: 'requested_lesson', ...booking }},
        { body: { action: 'notify', target: 'student',  via: 'email', after: remindOn.toISO(), message: 'confirm_lesson',   ...booking }},
        ...2 messages omitted...
    ]);

    await platform.env.bookings.sendBatch([
        { body: { action: 'notify', target: 'student',  via: 'email', after: now.toISO(),      message: 'requested_lesson', ...booking }},
        ...2 messages omitted...
      message: 'requested_lesson', ...booking }},
        { body: { action: 'notify', target: 'student',  via: 'email', after: remindOn.toISO(), message: 'confirm_lesson',   ...booking }},
        ...2 messages omitted...
    ]);

    await platform.env.bookings.sendBatch([
        { body: { action: 'notify', target: 'student',  via: 'email', after: now.toISO(),      message: 'requested_lesson', ...booking }},
        ...2 messages omitted...
      message: 'requested_lesson', ...booking }},
        { body: { action: 'notify', target: 'student',  via: 'email', after: remindOn.toISO(), message: 'confirm_lesson',   ...booking }},
        ...2 messages omitted...
    ]);

    await platform.env.bookings.sendBatch([
        { body: { action: 'notify', target: 'student',  via: 'email', after: now.toISO(),      message: 'requested_lesson', ...booking }},
        ...2 messages omitted...
      message: 'requested_lesson', ...booking }},
        { body: { action: 'notify', target: 'student',  via: 'email', after: remindOn.toISO(), message: 'confirm_lesson',   ...booking }},
        ...2 messages omitted...
    ]);

U

UNOwen•9/22/24, 9:49 AM

Message receiving:

    async queue(batch: MessageBatch<ActionMessage>, env: Env): Promise<void> {
        const now = DateTime.now();
        for (let message: Message<ActionMessage> of batch.messages) {
            const sendAfter = DateTime.fromISO(message.body.after);
            if (sendAfter > now) {
                const delaySeconds = Math.round(Math.min(
                    Duration.fromObject({ hours: 12}).as('seconds'),
                    sendAfter.diff(now).as('seconds')));
                message.retry({
                    delaySeconds: delaySeconds
                });
                console.info(`message ${message.id} DELAYED: ${JSON.stringify(message.body)} by ${delaySeconds}`);
            } else ...

    async queue(batch: MessageBatch<ActionMessage>, env: Env): Promise<void> {
        const now = DateTime.now();
        for (let message: Message<ActionMessage> of batch.messages) {
            const sendAfter = DateTime.fromISO(message.body.after);
            if (sendAfter > now) {
                const delaySeconds = Math.round(Math.min(
                    Duration.fromObject({ hours: 12}).as('seconds'),
                    sendAfter.diff(now).as('seconds')));
                message.retry({
                    delaySeconds: delaySeconds
                });
                console.info(`message ${message.id} DELAYED: ${JSON.stringify(message.body)} by ${delaySeconds}`);
            } else ...

    async queue(batch: MessageBatch<ActionMessage>, env: Env): Promise<void> {
        const now = DateTime.now();
        for (let message: Message<ActionMessage> of batch.messages) {
            const sendAfter = DateTime.fromISO(message.body.after);
            if (sendAfter > now) {
                const delaySeconds = Math.round(Math.min(
                    Duration.fromObject({ hours: 12}).as('seconds'),
                    sendAfter.diff(now).as('seconds')));
                message.retry({
                    delaySeconds: delaySeconds
                });
                console.info(`message ${message.id} DELAYED: ${JSON.stringify(message.body)} by ${delaySeconds}`);
            } else ...

    async queue(batch: MessageBatch<ActionMessage>, env: Env): Promise<void> {
        const now = DateTime.now();
        for (let message: Message<ActionMessage> of batch.messages) {
            const sendAfter = DateTime.fromISO(message.body.after);
            if (sendAfter > now) {
                const delaySeconds = Math.round(Math.min(
                    Duration.fromObject({ hours: 12}).as('seconds'),
                    sendAfter.diff(now).as('seconds')));
                message.retry({
                    delaySeconds: delaySeconds
                });
                console.info(`message ${message.id} DELAYED: ${JSON.stringify(message.body)} by ${delaySeconds}`);
            } else ...

U

UNOwen•9/22/24, 9:50 AM

And here's the relevant consumer configuration:

[[queues.consumers]]
queue = "litcondit-bookings"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 100
dead_letter_queue = "litcondit-bookings-dlq"

[[queues.consumers]]
queue = "litcondit-bookings"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 100
dead_letter_queue = "litcondit-bookings-dlq"

[[queues.consumers]]
queue = "litcondit-bookings"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 100
dead_letter_queue = "litcondit-bookings-dlq"

[[queues.consumers]]
queue = "litcondit-bookings"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 100
dead_letter_queue = "litcondit-bookings-dlq"

U

UNOwen•9/22/24, 9:53 AM

So far I've tried to work around this issue by having a "ping" message being sent to the queue every 5 minutes, which seems to help a little – the messages in my local region are now not getting stuck (I am located in Canada, and it seems to help in NA), but it doesn't prevent messages from getting into the limbo stage in EU (assuming there aren't more regions by which Queues are partitioned, which I don't know about).

UUNOwen Message receiving: ``` async queue(batch: MessageBatch<ActionMessage>, env:...

B

Bibi•9/22/24, 11:17 AM

Why don’t you use https://developers.cloudflare.com/queues/configuration/batching-retries/#delay-messages the delaySeconds property when creating the queue message?

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.

B

Bibi•9/22/24, 11:26 AM

Oh that has a limit of 12 hours too

B

Bibi•9/22/24, 11:36 AM

But you could acknowledge the message and send it again with delay instead of retrying

U

UNOwen•9/22/24, 11:47 AM

Re-sending messages might be a viable short-term workaround, but it would only function if sending messages is 100% reliable, and it doesn't seem like this "limbo" behavior is clearly intended – I am either misusing the retry API, or it has an edge case I am hitting. Re-sending the message would side-step the issue without documenting the details for future users (the "limbo" behavior is confusing and unlike other queues I worked with so far) or fixing the edge case.

B

boywonder350•9/22/24, 8:59 PM

I have the paid pro plan $25/month and the docs says queue consumers can have 15 min cpu time, but a worker I just created is timing out after 30s. If I try to update the cpu time via the dashboard, it says a maximum amount is 30s (30000ms). What is going on here?

B

boywonder350•9/22/24, 10:01 PM

^ Seems like this is a bug or something? I am on the "Standard" plan and the worker I am writing is only a queue consumer andn othing else

B

boywonder350•9/22/24, 10:04 PM

I hope this is not a docs. typo because I wrote a bunch of code with this presumption :/

Bboywonder350 I have the paid pro plan $25/month and the docs says queue consumers can have 15...

A

ajgeiss0702•9/22/24, 10:05 PM

i think you got them mixed up. They can have 15 minutes wall clock time. 30s cpu time

Aajgeiss0702 i think you got them mixed up. They can have 15 minutes __wall clock__ time. 30s...

B

boywonder350•9/22/24, 10:06 PM

A

ajgeiss0702•9/22/24, 10:07 PM

oh hm the docs are inconsistant

A

ajgeiss0702•9/22/24, 10:08 PM

apparently the 30s cpu time is what it actually is. Whether thats intentional or not, who knows

A

ajgeiss0702•9/22/24, 10:08 PM

wait where is that? The normal workers limits page just doesnt list queue consumers' cpu time limit

B

boywonder350•9/22/24, 10:09 PM

https://developers.cloudflare.com/workers/platform/pricing/#example-pricing-standard-usage-model

Cloudflare Docs

Pricing | Cloudflare Workers docs

Workers plans and pricing information.

A

ajgeiss0702•9/22/24, 10:10 PM

oh on the pricing page

A

ajgeiss0702•9/22/24, 10:10 PM

weird. hopefully someone from cf can help you later

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

Similar Threads

Similar Threads

Similar Threads