where each queue will have a different message body

Ooof2win2 where each queue will have a different message body

C

cloudkite•9/24/24, 9:44 PM

I had the same issue, ended up overriding the types to get that to work. Would be nice if it was supported out of the box:

interface TypedMessageBatch<Name extends string, Body> {
  readonly messages: readonly Message<Body>[];
  readonly queue: Name;
  retryAll(options?: QueueRetryOptions): void;
  ackAll(): void;
}

type MessageBatch =
  | TypedMessageBatch<"queue-1", { foo: number }>
  | TypedMessageBatch<"queue-2", { bar: number }>;
  

export default class extends WorkerEntrypoint<Env> {
  async queue(_batch: MessageBatch) {
      switch (batch.queue) {
        case "queue-1": {
          // message.body is typed
          break;
        }
    }
  }
}

interface TypedMessageBatch<Name extends string, Body> {
  readonly messages: readonly Message<Body>[];
  readonly queue: Name;
  retryAll(options?: QueueRetryOptions): void;
  ackAll(): void;
}

type MessageBatch =
  | TypedMessageBatch<"queue-1", { foo: number }>
  | TypedMessageBatch<"queue-2", { bar: number }>;
  

export default class extends WorkerEntrypoint<Env> {
  async queue(_batch: MessageBatch) {
      switch (batch.queue) {
        case "queue-1": {
          // message.body is typed
          break;
        }
    }
  }
}

interface TypedMessageBatch<Name extends string, Body> {
  readonly messages: readonly Message<Body>[];
  readonly queue: Name;
  retryAll(options?: QueueRetryOptions): void;
  ackAll(): void;
}

type MessageBatch =
  | TypedMessageBatch<"queue-1", { foo: number }>
  | TypedMessageBatch<"queue-2", { bar: number }>;
  

export default class extends WorkerEntrypoint<Env> {
  async queue(_batch: MessageBatch) {
      switch (batch.queue) {
        case "queue-1": {
          // message.body is typed
          break;
        }
    }
  }
}

interface TypedMessageBatch<Name extends string, Body> {
  readonly messages: readonly Message<Body>[];
  readonly queue: Name;
  retryAll(options?: QueueRetryOptions): void;
  ackAll(): void;
}

type MessageBatch =
  | TypedMessageBatch<"queue-1", { foo: number }>
  | TypedMessageBatch<"queue-2", { bar: number }>;
  

export default class extends WorkerEntrypoint<Env> {
  async queue(_batch: MessageBatch) {
      switch (batch.queue) {
        case "queue-1": {
          // message.body is typed
          break;
        }
    }
  }
}

O

oof2win2OP•9/24/24, 9:47 PM

oh thats great, thanks

T

tadhglewis•9/24/24, 9:48 PM

In combination of above, if you want runtime validation instead of doing assertions, you can use something like Zod with discriminated unions

const Body = z.union(
  z.object({ type: "A", data: {...} }),
  z.object({ type: "B", data: {...} })
);

try {
  msg = Body.parse(await req.json()); // msg is now guaranteed to match the `Body` schema
} catch (err) {
  return new Response("Invalid body", { status: 400 });
}

await env.QUEUE.send(msg);

const Body = z.union(
  z.object({ type: "A", data: {...} }),
  z.object({ type: "B", data: {...} })
);

try {
  msg = Body.parse(await req.json()); // msg is now guaranteed to match the `Body` schema
} catch (err) {
  return new Response("Invalid body", { status: 400 });
}

await env.QUEUE.send(msg);

const Body = z.union(
  z.object({ type: "A", data: {...} }),
  z.object({ type: "B", data: {...} })
);

try {
  msg = Body.parse(await req.json()); // msg is now guaranteed to match the `Body` schema
} catch (err) {
  return new Response("Invalid body", { status: 400 });
}

await env.QUEUE.send(msg);

const Body = z.union(
  z.object({ type: "A", data: {...} }),
  z.object({ type: "B", data: {...} })
);

try {
  msg = Body.parse(await req.json()); // msg is now guaranteed to match the `Body` schema
} catch (err) {
  return new Response("Invalid body", { status: 400 });
}

await env.QUEUE.send(msg);

This ensures the queue has only valid data

Ooof2win2 oh thats great, thanks

C

cloudkite•9/24/24, 10:18 PM

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

GitHub

Support discriminated unions for queue message batches · Issue #279...

Currently when a worker is a consumer for multiple queues the message body type does not depend on the name of the queue. Would be nice to make the queue name generic to support having discriminate...

A

ac•9/25/24, 12:40 PM

When do queue messages need to have a

body

body

property wrapping them and when do they not? The docs are not clear on this

A

ac•9/25/24, 12:47 PM

It seems like they might need to be wrapped in

body

body

when they are sent in a batch, but not wrapped when they are sent individually?

Ccloudkite https://github.com/cloudflare/workerd/issues/2795

O

oof2win2OP•9/25/24, 4:07 PM

thank you so much

B

Ben•9/25/24, 6:43 PM

I feel like I have a (somewhat) simple use case involving Durable Objects + Queues.

So my durable object is an orchestrator for my project. Every 10 seconds it runs a new "poll". It runs on a forever loop until my customer shuts it off.

During that "poll" I need to make 1000~ fetch requests to a third party api, and the durable object needs the responses for those 1000 fetch requests.

The problem is durable objects only support 6 open connections, so I have to iterate over the 1000 fetch requests, and it can take a while to iterate over that many async requests. I need these 1000 requests to resolve as quickly as possible.

My thinking is maybe queues could help with this? But my concern is queues won't resolve these requests fast enough since the consumer worker needs to iterate over the queues.

BBen I feel like I have a (somewhat) simple use case involving Durable Objects + Queu...

O

oof2win2OP•9/25/24, 10:32 PM

you'd probably just setup one queue, set max batch size to 100, each queue would iterate over the request and complete it. with worker concurrency you'd have 10 workers running and getting you the results

P

Pranshu Maheshwari•9/26/24, 1:39 PM

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

The major changes:
- Per queue maximum message throughput has increased from 400 => 5,000
- Maximum consumer concurrency has increased from 20 => 250
- You should see substantially lower latency for sending messages to a Queue

These changes have been applied already to all queues, new and existing. You don't need to take any action to enable the higher limits.

Thanks for your support during the beta!

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

M

Matt Silverlock•9/26/24, 3:08 PM

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

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)

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.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

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/

path (it has

cache

cache

and

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 Maheshwari•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 Maheshwari•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

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?

where each queue will have a different message body

Similar Threads

Similar Threads

Similar Threads