Hey folks, fresh Queues user here. I am having issues with queue messages not being processed after

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

U

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

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

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

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

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

Original message was deleted

I

Isaac McFadyen•9/23/24, 12:35 PM

Oy no leaking!

S

sharax•9/23/24, 2:42 PM

Is the queue service still unavailable?

S

sharax•9/23/24, 2:49 PM

When I connect the queue to the worker, the worker does not work at all.
It does not give any logical message

S

sharax•9/23/24, 2:49 PM

I get the warning ‘Error 1104 Ray ID: 8c782842b92136fa • 2024-09-23 05:32:58 UTC Script not found.’ What’s the issue?! How can I use this service?

S

sharax•9/23/24, 2:50 PM

There is a simple hello world script that looks like this after connecting

O

oof2win2•9/24/24, 4:45 PM

is there a way to differentiate between types for two different queues? like i have two attached to one worker, so i'd want to have different types for the message bodies based on the queue names - is this possible or do i need to do this manually

Ooof2win2 is there a way to differentiate between types for two different queues? like i h...

M

Matt Silverlock•9/24/24, 6:17 PM

Every message batch has the queue name - https://developers.cloudflare.com/queues/configuration/javascript-apis/#messagebatch

Cloudflare Docs

JavaScript APIs | Cloudflare Queues

Cloudflare Queues is integrated with Cloudflare Workers. To send and receive messages, you must use a Worker.

M

Matt Silverlock•9/24/24, 6:20 PM

Example here: https://developers.cloudflare.com/queues/reference/how-queues-works/#create-a-consumer-worker

O

oof2win2•9/24/24, 6:27 PM

yeah i know but can i restrict the typings based on the queue

O

oof2win2•9/24/24, 6:28 PM

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

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

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

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

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

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

Hey folks, fresh Queues user here. I am having issues with queue messages not being processed after

Similar Threads

Similar Threads

Similar Threads