I don't see wrangler.toml's equivalent of `[[queues.consumers]]` referenced in https://developers.cl

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

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

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

I don't see wrangler.toml's equivalent of `[[queues.consumers]]` referenced in https://developers.cl

Similar Threads

Similar Threads

Similar Threads