I am looking at queue api docs, there is no way to send something to queue outside of the worker. Am

SSameer Ali I am looking at queue api docs, there is no way to send something to queue outsi...

M

You would need to write a Worker: https://developers.cloudflare.com/queues/get-started/#write-your-producer-worker

Cloudflare Queues is a flexible messaging queue that allows you to queue messages for asynchronous processing. By following this guide, you will …

S

Sameer AliOP•10/4/23, 12:47 AM

Thanks

�

😈 Donkey 💫•10/4/23, 5:09 AM

my producer is simple:

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {

        // if req.method is not POST , reject
        if (req.method !== 'POST') {
            return new Response('Sorry', { status: 405 });
        }
        const body = await req.json();
        console.log(`got body: ${JSON.stringify(body)}`)
        const result = await env.MY_QUEUE.send({
            body: body,
            contentType: "json",
        });
        console.log(`sent message to queue: ${JSON.stringify(result)}`);
        return new Response(JSON.stringify({
            return_code: "1",
            return_message: "ok"
        }), {
            headers: { 'Content-Type': 'application/json' },
        });
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {

        // if req.method is not POST , reject
        if (req.method !== 'POST') {
            return new Response('Sorry', { status: 405 });
        }
        const body = await req.json();
        console.log(`got body: ${JSON.stringify(body)}`)
        const result = await env.MY_QUEUE.send({
            body: body,
            contentType: "json",
        });
        console.log(`sent message to queue: ${JSON.stringify(result)}`);
        return new Response(JSON.stringify({
            return_code: "1",
            return_message: "ok"
        }), {
            headers: { 'Content-Type': 'application/json' },
        });
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {

        // if req.method is not POST , reject
        if (req.method !== 'POST') {
            return new Response('Sorry', { status: 405 });
        }
        const body = await req.json();
        console.log(`got body: ${JSON.stringify(body)}`)
        const result = await env.MY_QUEUE.send({
            body: body,
            contentType: "json",
        });
        console.log(`sent message to queue: ${JSON.stringify(result)}`);
        return new Response(JSON.stringify({
            return_code: "1",
            return_message: "ok"
        }), {
            headers: { 'Content-Type': 'application/json' },
        });
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {

        // if req.method is not POST , reject
        if (req.method !== 'POST') {
            return new Response('Sorry', { status: 405 });
        }
        const body = await req.json();
        console.log(`got body: ${JSON.stringify(body)}`)
        const result = await env.MY_QUEUE.send({
            body: body,
            contentType: "json",
        });
        console.log(`sent message to queue: ${JSON.stringify(result)}`);
        return new Response(JSON.stringify({
            return_code: "1",
            return_message: "ok"
        }), {
            headers: { 'Content-Type': 'application/json' },
        });
    },
};

after make 1 request via PostMan

there's a lots triggering this function, make massive queues produced to the Queue

any of them, has same logs like below:

"logs": [
    {
      "message": [
        "got body: {\"data\":\"dat hart fd3 3\"}"
      ],
      "level": "log",
      "timestamp": 1696396087813
    },
    {
      "message": [
        "sent message to queue: undefined"
      ],
      "level": "log",
      "timestamp": 1696396087867
    }
  ]

"logs": [
    {
      "message": [
        "got body: {\"data\":\"dat hart fd3 3\"}"
      ],
      "level": "log",
      "timestamp": 1696396087813
    },
    {
      "message": [
        "sent message to queue: undefined"
      ],
      "level": "log",
      "timestamp": 1696396087867
    }
  ]

"logs": [
    {
      "message": [
        "got body: {\"data\":\"dat hart fd3 3\"}"
      ],
      "level": "log",
      "timestamp": 1696396087813
    },
    {
      "message": [
        "sent message to queue: undefined"
      ],
      "level": "log",
      "timestamp": 1696396087867
    }
  ]

"logs": [
    {
      "message": [
        "got body: {\"data\":\"dat hart fd3 3\"}"
      ],
      "level": "log",
      "timestamp": 1696396087813
    },
    {
      "message": [
        "sent message to queue: undefined"
      ],
      "level": "log",
      "timestamp": 1696396087867
    }
  ]

what's wrong with my code? or is this a bug? any helps?

�

😈 Donkey 💫•10/4/23, 6:07 AM

found out
it's because the routes endwith wildcard (star), without slashes rightbefore

( domain.com/a* => bug massive trigger)
( domain.com/a/* => works now)

M

marina.lipeck•10/4/23, 10:22 AM

hey guys! is there a way to increase the limit of queues? We used the request form (https://docs.google.com/forms/d/e/1FAIpQLSd_fwAVOboH9SlutMonzbhCxuuuOmiU1L_I5O2CFbXf_XXMRg/viewform) from the docs a couple of weeks ago, but never got any answer

Google Docs

Cloudflare Developer Platform: Limit Increase Request Form

Want to increase the limits of your current plan? Let us know what your requirements are here.

Mmarina.lipeck hey guys! is there a way to increase the limit of queues? We used the request fo...

M

Matt Silverlock•10/4/23, 11:07 AM

Which limit are you looking to increase?

MMatt Silverlock Which limit are you looking to increase?

M

marina.lipeck•10/4/23, 11:23 AM

this one https://developers.cloudflare.com/queues/platform/limits/#limits:~:text=Queues-,100%20per%20account,-1

Limits · Cloudflare Queues

Many of these limits will increase during Queues’ public beta period. Follow our changelog or join the #queues-beta channel in our Developer Discord …

Mmarina.lipeck this one https://developers.cloudflare.com/queues/platform/limits/#limits:~:text...

M

Matt Silverlock•10/4/23, 11:35 AM

I can't see which request is yours: can you share your account ID here? I'll increase it to 10,000.

MMatt Silverlock I can't see which request is yours: can you share your account ID here? I'll inc...

M

marina.lipeck•10/4/23, 11:37 AM

thanks! I just received the email that increased is done

Mmarina.lipeck thanks! I just received the email that increased is done

M

Matt Silverlock•10/4/23, 11:40 AM

Great! There were two outstanding asks, so was hoping one was yours

S

switz•10/4/23, 2:12 PM

it says you can set it from 0-60, but if you set it to 0 it defaults back to 5

S

switz•10/4/23, 2:12 PM

is 0 possible?

S

switz•10/4/23, 2:13 PM

no delay for batching

S

switz•10/4/23, 2:14 PM

1 second is fine for my use-cases, but in an ideal world things would process as they come in – perhaps as low as 50ms or 200ms if a batch is necessary?

S

switz•10/4/23, 2:15 PM

but yeah, processing 1 at a time is also ok by me, if it means its faster

Original message was deleted

N

nclevenger•10/4/23, 3:44 PM

I just realized I was seeing this behavior too ... and yes, in our case, there are certain things that are very latency sensitive so at low volumes, immediate is best, but as volumes scale then they will naturally batch ... but yes, 50ms or 100ms would be ideal for those ... but we had been trying to set to 0

V

vikiival | koda.art•10/5/23, 10:24 AM

Hello,
has anyone tried honojs or other framework for queue worker?

K

kian•10/5/23, 10:25 AM

Hono is purely for the

fetch

fetch

handler so if you mean you want

fetch

fetch

w/ Hono and

queue

queue

handlers in the same project, then just do it like this

export default {
  fetch: app.fetch,
  async queue(...) { ... }
}

export default {
  fetch: app.fetch,
  async queue(...) { ... }
}

export default {
  fetch: app.fetch,
  async queue(...) { ... }
}

export default {
  fetch: app.fetch,
  async queue(...) { ... }
}

Kkian Hono is purely for the `fetch` handler so if you mean you want `fetch` w/ Hono a...

V

vikiival | koda.art•10/5/23, 10:26 AM

Thank you very much ferrisBongoHyper

Kkian Hono is purely for the `fetch` handler so if you mean you want `fetch` w/ Hono a...

V

vikiival | koda.art•10/5/23, 10:39 AM

Offtopic but same also works for

scheduled

scheduled

worker right?

A

Adi//•10/6/23, 6:00 AM

Hi just signed up for the paid worker plan
Hi what does this exactly mean in limits?
I have a queue with batching 1 message,1 second
Does that mean my average requests per second can be max 10?

This is the number of "starts" per second right? Not the number of concurrent queues?

Screenshot_2023-10-06_at_11.28.06_AM.png

AAdi//Hi just signed up for the paid worker plan Hi what does this exactly mean in li...

M

Matt Silverlock•10/6/23, 9:07 AM

It is linked directly to https://developers.cloudflare.com/queues/learning/consumer-concurrency/#enable-concurrency

Consumer concurrency · Cloudflare Queues

Consumer concurrency allows a consumer Worker processing messages from a queue to automatically scale out horizontally to keep up with the rate that …

M

Matt Silverlock•10/6/23, 4:22 PM

If you're using Queues, you can now create up to 10,000 queues per account: https://developers.cloudflare.com/queues/changelog/#more-queues-per-account---up-to-10000

Changelog · Cloudflare Queues

Queue consumers can now scale to 20 concurrent invocations (per queue), up from 10. This allows you to scale out and process higher throughput queues …

U

Unsmart•10/6/23, 4:24 PM

Oh nice 20 concurrent invokes also yesterday / 2 days ago Pog

U

Unsmart•10/6/23, 4:26 PM

What timezone are the changelogs in for it to says the queue count is the 7th? Or was that just the planned date and wasnt changed when it happened a day early

UUnsmart What timezone are the changelogs in for it to says the queue count is the 7th? O...

M

Matt Silverlock•10/6/23, 5:09 PM

Day early

UUnsmart Oh nice 20 concurrent invokes also yesterday / 2 days ago <:Pog:9027562436916839...

M

Matt Silverlock•10/6/23, 5:10 PM

And yep: queues will auto-scale to 20 concurrent consumers now. This will continue to get higher, but there’s more work to do here, and I’d like to get message delays in place to go alongside so there’s more control over throughput.

F

forslunds•10/6/23, 6:46 PM

Sweet! 1of2 problems solved to migrate from google

M

Maxl•10/6/23, 8:22 PM

I discovered a bug related to the queue

contentType

contentType

option [1] and dead letter queues [2].
When

contentType

contentType

is set to anything other than

"v8"

"v8"

and the queue event ends up in the DLQ, a consumer worker attached to that DLQ cannot handle it.

// Any contentType other than "v8" cannot be handled in the DLQ
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "json" });

// This is handled fine by the DLQ consumer worker
env.MY_QUEUE.send({ foo: "bar" });
// This is also fine
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "v8" });

// Any contentType other than "v8" cannot be handled in the DLQ
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "json" });

// This is handled fine by the DLQ consumer worker
env.MY_QUEUE.send({ foo: "bar" });
// This is also fine
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "v8" });

// Any contentType other than "v8" cannot be handled in the DLQ
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "json" });

// This is handled fine by the DLQ consumer worker
env.MY_QUEUE.send({ foo: "bar" });
// This is also fine
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "v8" });

// Any contentType other than "v8" cannot be handled in the DLQ
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "json" });

// This is handled fine by the DLQ consumer worker
env.MY_QUEUE.send({ foo: "bar" });
// This is also fine
env.MY_QUEUE.send({ foo: "bar" }, { contentType: "v8" });

The logs of the DLQ consumer worker show for any such non-v8 event:

"Unable to deserialize cloned data due to invalid or unsupported version."

"Unable to deserialize cloned data due to invalid or unsupported version."

Here's a minimal example of a producer-consumer worker to reproduce the issue:

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env): Promise<Response> {
        // Any contentType other than "v8" cannot be handled in the DLQ
        await env.MY_QUEUE.send({ foo: 'bar' }, { contentType: 'json' });
        return new Response('Sent message to the queue');
    },

    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        // Due to this error, the event will end up in the dead letter queue
        throw new Error('Foobar');
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env): Promise<Response> {
        // Any contentType other than "v8" cannot be handled in the DLQ
        await env.MY_QUEUE.send({ foo: 'bar' }, { contentType: 'json' });
        return new Response('Sent message to the queue');
    },

    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        // Due to this error, the event will end up in the dead letter queue
        throw new Error('Foobar');
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env): Promise<Response> {
        // Any contentType other than "v8" cannot be handled in the DLQ
        await env.MY_QUEUE.send({ foo: 'bar' }, { contentType: 'json' });
        return new Response('Sent message to the queue');
    },

    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        // Due to this error, the event will end up in the dead letter queue
        throw new Error('Foobar');
    },
};

export interface Env {
    MY_QUEUE: Queue;
}

export default {
    async fetch(req: Request, env: Env): Promise<Response> {
        // Any contentType other than "v8" cannot be handled in the DLQ
        await env.MY_QUEUE.send({ foo: 'bar' }, { contentType: 'json' });
        return new Response('Sent message to the queue');
    },

    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        // Due to this error, the event will end up in the dead letter queue
        throw new Error('Foobar');
    },
};

And a minimal DLQ consumer:

export default {
    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        console.log(`Events received: ${JSON.stringify(batch.messages)}`);
    },
};

export default {
    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        console.log(`Events received: ${JSON.stringify(batch.messages)}`);
    },
};

export default {
    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        console.log(`Events received: ${JSON.stringify(batch.messages)}`);
    },
};

export default {
    async queue(batch: MessageBatch<Error>, env: Env): Promise<void> {
        console.log(`Events received: ${JSON.stringify(batch.messages)}`);
    },
};

Relevant docs:
[1] https://developers.cloudflare.com/queues/platform/javascript-apis/#queuescontenttype
[2] https://developers.cloudflare.com/queues/learning/batching-retries/#dead-letter-queues

MMaxl I discovered a bug related to the queue `contentType` option [1] and dead letter...

M

Matt Silverlock•10/6/23, 11:35 PM

Thanks for the report. Capturing internally so this isn’t lost.

MMatt Silverlock And yep: queues will auto-scale to 20 concurrent consumers now. This will contin...

A

Adi//•10/7/23, 7:24 AM

hi just confirming this means i can have 20 consumers running parallely right?
If my batch limit is 1 that means If my request takes 1 second to handle my capacity is 20 per second?

A

Adi//•10/7/23, 7:25 AM

(and if I want a higher throughput I need to start batching and then do a promise.All inside my consumer?)

AAdi//hi just confirming this means i can have 20 consumers running parallely right? ...

M

Matt Silverlock•10/9/23, 5:53 PM

I’d strongly recommend batching (and keeping your batch wait time low if you have latency needs) - processing only one message per invocation will never scale

M

malone•10/10/23, 10:11 PM

I sometimes get this error, which I am pretty sure is coming from Queues

Unknown Event - Exception Thrown @ 10/10/2023, 3:10:17 PM
✘ [ERROR]   Error: Unable to deserialize cloned data due to invalid or unsupported version.

Unknown Event - Exception Thrown @ 10/10/2023, 3:10:17 PM
✘ [ERROR]   Error: Unable to deserialize cloned data due to invalid or unsupported version.

I added a dead letter queue and started seeing this. Is there some way I should specify what type my queue data is? Or is this unrelated?

K

kian•10/10/23, 10:14 PM

It's been reported above (queues) and logged internally

Kkian It's been reported above (https://discord.com/channels/595317990191398933/100869...

M

malone•10/10/23, 10:17 PM

aha, so omitting json makes it work? i was just assuming i should use json but happy to remove it if it is better not to

MMatt Silverlock I’d strongly recommend batching (and keeping your batch wait time low if you hav...

A

Adi//•10/11/23, 11:30 AM

yes just did this earlier started batching, and then doing a promise.all inside the queue consumer to resolve it

D

divby0•10/15/23, 9:22 AM

Since pages functions can not be consumers, would my naive approach of having a worker-consumer that does nothing else than call and await my api that's hosted on cloudflare pages work?

I have my main application on Pages if that isn't clear from the above

Ddivby0 Since pages functions can not be consumers, would my naive approach of having a ...

H

Hello, I’m Allie!•10/15/23, 9:23 AM

Yeah, that should work fine

HHello, I’m Allie!Yeah, that should work fine

D

divby0•10/15/23, 9:24 AM

and the best approach is to simply call the api endpoint via "internet" and cloudflare figures out it doesn't need to really call outside their network or is there anything I need to do to make the latency small (cause it's probably the same isolate-host that will run both things)

Ddivby0 and the best approach is to simply call the api endpoint via "internet" and clou...

H

Hello, I’m Allie!•10/15/23, 9:27 AM

Oh wait, like that. Why not just put the consumer logic in the Worker?

H

Hello, I’m Allie!•10/15/23, 9:27 AM

Then your Functions are lighter too

HHello, I’m Allie!Oh wait, like that. Why not just put the consumer logic in the Worker?

D

divby0•10/15/23, 9:29 AM

I was thinking about this as well - but at the moment I have one main pages app that's built on the Remix framework. I am hesitant to have 2 big projects with frameworks that need deploying and active maintentance etc. Rather would have one code base, one CI/CD etc.

But maybe the seperate approach would be better. I mean it's probably faster and maybe I can even split functionality at a good point where workers do x features and pages does y features... Kind of like having micro services for some things in seperate workers.

D

divby0•10/15/23, 9:31 AM

for the "reverse-proxy" worker, I'd probably just proxy the request with a plain js worker. If I want actual functionality on the worker, I definitely want it to be a remix project as well. Not sure if the set up is worth it. + I'd need to elevate the database scheme up somehow so they are accessible by both projects...

Do you know any resources about people that have done things like this or point me into a directioN?

S

Stefan•10/15/23, 2:14 PM

Hi, i'm currently building a solidstart application which will be deployed to cloudflare pages, I need to be able to send out verification emails when a user signs up, which i'm using aws ses for, and was wondering about general reliability and stability of cloudflare queues, and if it was suitable for production usage, and if there was anything to be aware of before proceeding?

V

vassbence•10/15/23, 2:59 PM

Any plans for adding a configurable delay for items? Similar to SQS

V

vassbence•10/15/23, 3:01 PM

For example in my SaaS when an user creates an order but does not finish the payment in 10 minutes I need to cancel the order and update stock numbers etc. Currently I'm doing this by running a CRON jon every minute and reading items from the DB. Problem is that it would be a lot more scalable if I could use a Queue where the items would start off with the an initial delay and once that goes away that would mean that the item can be consumed from the queue ( = cancel the order). Having individial consumert instead of a single CRON job worker would be a lot cleaner.

M

malone•10/16/23, 1:46 AM

are there any plans to have priority built into the queue?

M

malone•10/16/23, 4:59 AM

how many queue consumers run in parallel, if any?

I am looking at queue api docs, there is no way to send something to queue outside of the worker. Am

Similar Threads