Folks: in case you haven't seen the announcement yet, Queues is officially GA. The major changes:

PPranshu Maheshwari Folks: in case you haven't seen the announcement yet, Queues is officially GA. ...

M

The Cloudflare Blog

Builder Day 2024: 18 big updates to the Workers platform

To celebrate Builder Day 2024, we’re shipping 18 updates inspired by direct feedback from developers building on Cloudflare. This includes new capabilities, like running evals with AI Gateway, beta products like Queues and Vectorize graduating to GA, massive speed improvements to KV and Workers AI, pricing simplifications, and bringing key featu...

M

minty•9/26/24, 9:21 PM

"Could not acknowledge messages"

Getting this error when trying to clear a dead-letter queue

I

irajsavadlo•9/27/24, 10:11 AM

Ok

Mm0c Short question, I am slightly confused of how many messages in my dlq backlog ar...

M

Matt Silverlock•9/27/24, 12:31 PM

@Pranshu Maheshwari

PPranshu Maheshwari Folks: in case you haven't seen the announcement yet, Queues is officially GA. ...

0

0x146231489231923•9/27/24, 1:56 PM

5000 is so small if being used in a high transaction system?

S

square•9/28/24, 5:21 PM

Is there any plant to increase this message retention period ? (https://developers.cloudflare.com/queues/platform/limits/#:~:text=retention%20period%203-,4%20days,-(96%20hours)) 4 days is too low to build any serious application. for e.g. AWS SQS has a max retention period of 14 days. I thought it would be also increased during GA announcement.

Z

Zack•9/28/24, 6:21 PM

Yea, kind of agreed on the above. Would be nice to be able to set 7 days at least. With 4 days, if you receive a failure on Friday, you basically only have Monday to pick it up and replay

P

princeMN•9/28/24, 10:12 PM

#queues

A

an•9/29/24, 2:08 AM

Can I let a queue message run in a specific colo/ region? any idea to achieve this

Ssquare Is there any plant to increase this message retention period ? (https://develope...

S

Sid•9/29/24, 2:09 AM

4 would be the default if left unset. The plan is to allow up to 14 days.

S

Sid•9/29/24, 2:09 AM

Let me fix that doc up real quick

CCetin /⧸/ Elefunc <λ>Are Queues down? They stopped working for me: consumer handlers do not get trigg...

T

tadhglewis•9/30/24, 4:03 AM

Working fine for me

P

pat•9/30/24, 9:23 AM

Hey fellows, I've got a really simple queues and d1 setup that is deployed as a worker. It is a consumer and a producer of the job that it handles. It basically uses the default wrangler.toml and I followed the tutorials on the respective page to set it up. The fetch just publishes a single event, the queue just logs it

export default {
    async fetch(request: Request<unknown, IncomingRequestCfProperties<unknown>>, env: Env, ctx: ExecutionContext): Promise<Response> {
        await env.aramDataCollectionQueue.send({
            riot_puuid: "secret",
            region: RegionGroups.SEA
        })

        return new Response("Success");
    },

    queue(batch: MessageBatch<CollectAramDataJob>, env: Env, ctx: ExecutionContext): void | Promise<void> {
        console.log(batch);
    }
} satisfies ExportedHandler<Env, CollectAramDataJob>

export default {
    async fetch(request: Request<unknown, IncomingRequestCfProperties<unknown>>, env: Env, ctx: ExecutionContext): Promise<Response> {
        await env.aramDataCollectionQueue.send({
            riot_puuid: "secret",
            region: RegionGroups.SEA
        })

        return new Response("Success");
    },

    queue(batch: MessageBatch<CollectAramDataJob>, env: Env, ctx: ExecutionContext): void | Promise<void> {
        console.log(batch);
    }
} satisfies ExportedHandler<Env, CollectAramDataJob>

export default {
    async fetch(request: Request<unknown, IncomingRequestCfProperties<unknown>>, env: Env, ctx: ExecutionContext): Promise<Response> {
        await env.aramDataCollectionQueue.send({
            riot_puuid: "secret",
            region: RegionGroups.SEA
        })

        return new Response("Success");
    },

    queue(batch: MessageBatch<CollectAramDataJob>, env: Env, ctx: ExecutionContext): void | Promise<void> {
        console.log(batch);
    }
} satisfies ExportedHandler<Env, CollectAramDataJob>

export default {
    async fetch(request: Request<unknown, IncomingRequestCfProperties<unknown>>, env: Env, ctx: ExecutionContext): Promise<Response> {
        await env.aramDataCollectionQueue.send({
            riot_puuid: "secret",
            region: RegionGroups.SEA
        })

        return new Response("Success");
    },

    queue(batch: MessageBatch<CollectAramDataJob>, env: Env, ctx: ExecutionContext): void | Promise<void> {
        console.log(batch);
    }
} satisfies ExportedHandler<Env, CollectAramDataJob>

I am trying to develop locally with

bunx wrangler dev

bunx wrangler dev

bunx wrangler dev

bunx wrangler dev but when I go to the url it runs on it just hangs for a while with no error output. After a reasonable amount of time it will start lagging out my computer and starts spamming to console

[wrangler:inf] GET / 500 Internal Server Error (78202ms)

[wrangler:inf] GET / 500 Internal Server Error (78202ms)

[wrangler:inf] GET / 500 Internal Server Error (78202ms)

[wrangler:inf] GET / 500 Internal Server Error (78202ms)

I am very new to cloudflare in general but I couldnt find anything in the docs

M

Mateus Eduardo•9/30/24, 6:44 PM

hello help me, even though the batch size is 1 when I trigger a message in the queue with delay seconds of 5 minutes. After this time, this message is not sent to my consumer. Is it only triggered when I add another message to the queue?

F

flof.fly•9/30/24, 9:27 PM

How do I create a queue for a test env worker, and another queue for another worker in prod?
I have a bit of confusion on how multi-env works in wrangler

K

kian•9/30/24, 9:28 PM

[[env.???.queues.consumers]]

[[env.???.queues.consumers]]

[[env.???.queues.consumers]]

[[env.???.queues.consumers]]/

[[env.???.queues.producers]]

[[env.???.queues.producers]]

[[env.???.queues.producers]]

[[env.???.queues.producers]]

F

flof.fly•9/30/24, 9:28 PM

Nocapping

F

flof.fly•9/30/24, 9:28 PM

thanks a lot kian

P

pao ramen•10/8/24, 9:47 AM

Hello! I have a very misterious bug with queues. I have a job that iterates a batch of 3 messages, but it only iterates 2 messages and then it never gets to the third:

export async function queue(batch, env) {
    console.log({ messages: JSON.stringify(batch.messages) }) // This logs an array of 3 objects

    for (const msg of batch.messages) {
        console.log({ message: JSON.stringify(msg) }) // This only logs twice

        try {
            // ...do things
            msg.ack()
        } catch (e) {
            console.error(`adjustUser error: ${e}`) // This never logs
            msg.retry()
        }
    }
}

export async function queue(batch, env) {
    console.log({ messages: JSON.stringify(batch.messages) }) // This logs an array of 3 objects

    for (const msg of batch.messages) {
        console.log({ message: JSON.stringify(msg) }) // This only logs twice

        try {
            // ...do things
            msg.ack()
        } catch (e) {
            console.error(`adjustUser error: ${e}`) // This never logs
            msg.retry()
        }
    }
}

export async function queue(batch, env) {
    console.log({ messages: JSON.stringify(batch.messages) }) // This logs an array of 3 objects

    for (const msg of batch.messages) {
        console.log({ message: JSON.stringify(msg) }) // This only logs twice

        try {
            // ...do things
            msg.ack()
        } catch (e) {
            console.error(`adjustUser error: ${e}`) // This never logs
            msg.retry()
        }
    }
}

export async function queue(batch, env) {
    console.log({ messages: JSON.stringify(batch.messages) }) // This logs an array of 3 objects

    for (const msg of batch.messages) {
        console.log({ message: JSON.stringify(msg) }) // This only logs twice

        try {
            // ...do things
            msg.ack()
        } catch (e) {
            console.error(`adjustUser error: ${e}`) // This never logs
            msg.retry()
        }
    }
}

This is my wrangler configuration

[[queues.producers]]
queue = "adjust-user"
binding = "ADJUST_USER_QUEUE"

[[queues.consumers]]
queue = "adjust-user"
max_batch_size = 10
max_batch_timeout = 5
retry_delay = 600
dead_letter_queue = "failed-adjust-user"

[[queues.producers]]
queue = "adjust-user"
binding = "ADJUST_USER_QUEUE"

[[queues.consumers]]
queue = "adjust-user"
max_batch_size = 10
max_batch_timeout = 5
retry_delay = 600
dead_letter_queue = "failed-adjust-user"

[[queues.producers]]
queue = "adjust-user"
binding = "ADJUST_USER_QUEUE"

[[queues.consumers]]
queue = "adjust-user"
max_batch_size = 10
max_batch_timeout = 5
retry_delay = 600
dead_letter_queue = "failed-adjust-user"

[[queues.producers]]
queue = "adjust-user"
binding = "ADJUST_USER_QUEUE"

[[queues.consumers]]
queue = "adjust-user"
max_batch_size = 10
max_batch_timeout = 5
retry_delay = 600
dead_letter_queue = "failed-adjust-user"

Any clue?

S

Saqib•10/9/24, 3:08 AM

Hey folks, I am trying to do some processing on queues, surprisingly they timeout after 60s inside the consumer. On the paid plan, currently testing locally. Any thoughts? It just stops processing no error, nothing.

T

toadfans•10/9/24, 9:14 AM

hello, is queue down now?

{

            "name": "Error",
            "message": "Queue send failed: Internal Server Error",
            "stack": "Error: Queue send failed: Internal Server Error\n    at async Object.mutation (bundledWorker-0.9232964110947641.mjs:9562:3)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:28094:17)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:14535:28)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:8361:42)"
          }

{

            "name": "Error",
            "message": "Queue send failed: Internal Server Error",
            "stack": "Error: Queue send failed: Internal Server Error\n    at async Object.mutation (bundledWorker-0.9232964110947641.mjs:9562:3)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:28094:17)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:14535:28)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:8361:42)"
          }

{

            "name": "Error",
            "message": "Queue send failed: Internal Server Error",
            "stack": "Error: Queue send failed: Internal Server Error\n    at async Object.mutation (bundledWorker-0.9232964110947641.mjs:9562:3)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:28094:17)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:14535:28)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:8361:42)"
          }

{

            "name": "Error",
            "message": "Queue send failed: Internal Server Error",
            "stack": "Error: Queue send failed: Internal Server Error\n    at async Object.mutation (bundledWorker-0.9232964110947641.mjs:9562:3)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:28094:17)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:14535:28)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async r (bundledWorker-0.9232964110947641.mjs:4210:92)\n    at async o (bundledWorker-0.9232964110947641.mjs:4128:13)\n    at async Object.handler (bundledWorker-0.9232964110947641.mjs:8361:42)"
          }

requestId: 8cfd3fdf8f3984b7

L

Lance•10/9/24, 1:53 PM

Hi, anyone encounters similar issue that queue's backlog just grows and no message is delivered to worker handler? Nothing changed from code side. It just suddenly happened a few hours ago.

S

Sam•10/13/24, 1:04 PM

Any update on when queues are coming to remote mode

https://github.com/cloudflare/workerd/issues/855

or is it possible to run queues locally and my other bindings remotely?

GitHub

🐛 BUG: `wrangler dev --remote` fails with Error 1104, Script not fo...

What version of Wrangler are you using? 2.6.2 What operating system are you using? Windows Describe the Bug Add a queue binding in your wrangler.toml, in my case: [[queues.producers]] queue = &quot...

A

ac•10/21/24, 1:50 PM

Hi, I am having issues using a Pages deployment as a queue producer. It doesn't throw any error, and the service binding for the Queue is there for the Pages app, but the queue never gets a message even though the enqueueing code is definitely getting called. I did notice that on the queue, the "Producer" shows a production and a preview deployment but they just have a generic

pages-worker-###-production

pages-worker-###-production

pages-worker-###-production

pages-worker-###-production format and aren't using the name of my Pages app

D

dangoldin•10/22/24, 8:37 PM

Hopefully a quick question - I'm doing some local dev and wanted to confirm my queues are set up properly but I don't see anything queue related in the

.wrangler/state/v3/

.wrangler/state/v3/

.wrangler/state/v3/

.wrangler/state/v3/ path (it has

cache

cache

cache

cache and

d1

d1

d1

d1). Should there be anything created there?

D

dangoldin•10/22/24, 8:39 PM

I have a Remix app that's published to a queue and then a separate worker to consume so trying to figure out why the worker isn't seeing the messages being published

AAhmad Awais whenever we add a Queue — it becomes a bottleneck for the entire platform instea...

J

John Spurlock•10/22/24, 9:35 PM

+1, even a notion like sqs's message-group-id as a scaling worker consumer hint, not for fifo (since cf doesn't do fifo)

AAhmad Awais whenever we add a Queue — it becomes a bottleneck for the entire platform instea...

P

Pranshu MaheshwariOP•10/23/24, 2:01 PM

thanks for the feedback here! very detailed & agreed that this is a pain point with queues today

we're thinking about ways to improve this. our plan right now is to add the ability to connect multiple consumers to a single queue. you'll be able to specify partition keys to route specific messages to specific consumers. and ideally this can be dynamic, so you don't have to specify all the rules ahead of time. this project is still in the design phase though, so it's too soon to share timelines.

dynamically creating queues doesn't work well today, alas! rather than 1 queue per user, would you be able to setup a few queues as high-priority queues, and setup separate low priority queues? by default, messages get routed to the high priority queues. if a user uploads 10k files at once, you route those messages to the low priority queues so that they don't block each other?

B

Brett Willis•10/24/24, 11:29 AM

Any plans to support pushing to Queues from an external platform direct via the rest API (not via an intermediate Worker)?

P

Pranshu MaheshwariOP•10/24/24, 3:07 PM

We published a deep dive on the Queues architecture today!
https://blog.cloudflare.com/how-we-built-cloudflare-queues/

The Cloudflare Blog

Durable Objects aren't just durable, they're fast: a 10x speedup fo...

Learn how we built Cloudflare Queues using our own Developer Platform and how it evolved to a geographically-distributed, horizontally-scalable architecture built on Durable Objects. Our new architecture supports over 10x more throughput and over 3x lower latency compared to the previous version.

S

Saqib•10/24/24, 6:03 PM

I have a queue with batch of size 50 and timeout of 10s.

Few questions:
1- If I do

.send

.send

.send

.send will msg be processed as soon as it arrives or it will wait for 10s before processing?
2- If a batch of 50 msgs is being processed by consumer and it takes longer than 10s, what happens next, if there are more messages waiting to be processed.
2.1- Will a new batch be delivered to the consumer? If so, will it make a new instance of the consumer or same isntance of consumer will process it.
2.2 Will new batch wait till the processing of previous batch is complete?
3- There is something related to concurrency. But it seems like it is never triggered. How can I verify if it is being triggered?

For now it seems like all processing is sequential. It becomes a bottleneck very quickly if mutliple users are using it.

A

ajgeiss0702•10/26/24, 12:37 AM

messages in my queue, even if they dont have a delay attached to them, are often not being delivered until 20-30 minutes after they are queues (sometimes more, the most ive seen so far is 1h 20m). any idea why that might be?

A

ajgeiss0702•10/26/24, 12:37 AM

and ones that do have a delay attached to them are 20-30 minutes later than they are supposed to be

A

ajgeiss0702•10/26/24, 12:40 AM

it looks like it started happening on october 23rd, around 9pm UTC

U

ueryooo•10/26/24, 1:33 PM

Can R2 Event Notifications to Queue be tested in local development?

M

meera_datey•10/27/24, 2:17 AM

Hello - I have a Queue with polling client running behind firewall. What is appropriate interval time to poll?

I also want to understand cost for this continuous polling.

M

markymark•10/28/24, 8:43 PM

Hello! I would be interested in FIFO ordering and also interested to know if you are going to support something like AWS SQS Message Group ID?

I tried to send an email to queues@cloudflare.com as specified in the docs (https://developers.cloudflare.com/queues/configuration/javascript-apis/#messagebatch) but that bounced.

Thanks!

J

jordanfinners•10/30/24, 6:27 PM

Hello, is there an issue with queues at the moment? I've got messages being reported as being in my deadletter, however I can't see them when listing messages and nothing else is reading from the queue

A

ac•10/30/24, 6:36 PM

Is it possible to add multiple messages to a queue from the dashboard? Seems like no matter what JSON I put in there (even trying to replicate a message batch) it doesn't work

S

SteffTek•10/31/24, 6:38 AM

Small Question about Concurrency - if i publish a message with the same content twice, will only one be processed? Example: i have a worker that publishes an event to tell the queue to update an object in an database, if this step is triggered twice, and two queue workers will be invoke, this would possible create a race condition

A

Alex•10/31/24, 2:17 PM

Yes, each message is a message so both will be processed possibly concurrently if you have a busy queue. You can disable concurrency though if you need to process them more controlled. See: https://developers.cloudflare.com/queues/configuration/consumer-concurrency/

AAlex Yes, each message is a message so both will be processed possibly concurrently i...

S

SteffTek•10/31/24, 2:27 PM

thx, yeah i have concurrency disabled currently, just hoped i overlooked something. we have a really busy queue and without concurrency it takes a while. i might add a second queue, for events where race conditions are a non-issue

Folks: in case you haven't seen the announcement yet, Queues is officially GA. The major changes:

Similar Threads

Similar Threads

Similar Threads