I discovered a bug related to the queue `contentType` option [1] and dead letter queues [2]. When `c

I discovered a bug related to the queue

contentType

contentType

contentType

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

contentType

contentType

contentType

contentType is set to anything other than

"v8"

"v8"

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

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

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

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?

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

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

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?

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

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

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

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

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

Yeah, that should work fine

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

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

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

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

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?

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.

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?

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?

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

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

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.

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

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

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

how many queue consumers run in parallel, if any?

CCharlie What would this feature look like to you?

malone•10/16/23, 3:04 PM

maybe as simple as having a number for the priority when queueing a message and the highest priority messages are dequeued first.

i hit an issue where one noisy user caused a backlog in my queue of multiple hours and im currently implementing a priority queue and expiring messages on top of the cloudflare queue. would be great if they were native features

forslunds•10/16/23, 7:02 PM

This would also be able to solve with a peogrammaticly create-one-queue-per-user

CCharlie While I may be a little biased as the PM for Queues, I'd say its pretty stable a...

Stefan•10/16/23, 7:15 PM

Haha, that's what I was hoping to hear! Cheers!

divby0•10/18/23, 6:33 AM

https://developers.cloudflare.com/queues/learning/local-development/#known-issues

This is a blocker. I have a Remix full stack cloudflare pages app that produces messages and a worker that consumes them. They way my project is set up, the worker doesn't have a ORM or any framework, it's just a helper so I can have a queue consumer.

That means I can't put my

queue

queue

method into my pages app and I can't put the producing code that execute sendBatchsendBatch into the worker either. Is there no solution for this?

arbfay•10/18/23, 12:48 PM

Error: Queue send failed: Service Temporarily UnavailableError: Queue send failed: Service Temporarily Unavailable ??

arbfay•10/18/23, 12:51 PM

What is the service level for queues? Send failure rates?

CCharlie Its on the roadmap 😄

vassbence•10/18/23, 6:29 PM

Is it a goal before the product leaves beta?

MMatt Silverlock This is coming (HTTP Pull) in the future.

switz•10/18/23, 7:15 PM

any idea when this is coming, it's basically the only thing keeping me on SQS

iustin.pericica•10/18/23, 8:08 PM

Hi there,
We have a project based on cloudflare workers & queues. We currently have 1 queue with this config ( batch size:50, wait time: 3 sec, msg retries: 1, max concurrency: 10). The workers send batches of messages to the queue and sometimes it gets overloaded thus we expected auto scaling to work, but we've not yet seen auto scaling(concurrency) working there.
For some reason the maximum reference in the metrics graph is 1:
Thanks!

CCharlie Can you help me understand this a bit more? Are multiple Worker scripts a possib...

divby0•10/19/23, 6:23 AM

Of course : )

So basically I want to locally debug the queue producer and queue consumer. The queue producer is a cloudflare pages app, built on Remix. The queue consumer is a worker with plain typescript, no framework.

I kept the worker that simple intentionally because it doesn't need database access or routing or whatever, it's basically a proxy for queue consumption (because cloudflare pages apps can't consume from a queue yet and other limitations).

If it was possible, I would use only the pages app as the producer and consumer. But my solution with having a separate consumer and producer is really not even what I want, but cloudflare's restriction of me not being able to use pages for everything means I need to split queue production and consumption.

I can't put the queue consumption into the pages app because pages apps can't be consumers. I can't put the queue production into the worker script because I don't have access to database and everything there.

Currently I just have a route on the pages app that I use as a workaround, but not being able to locally test pages<->workers communication via a queue is not nice

GuillaumeLakano•10/19/23, 7:41 AM

Hi! In the Known Issues for Local Dev, the documentation explain the producer and consumer need to be on the same worker.
How to do when multiple workers need to be a producer, but we only need one consumer ? After each line changed of our consumer code, we need to copy it to our ~20 workers that are producers ?
I known it's temporary, but it's really not useful to start to works with Queue. Do you have any suggestion to make it more useful please?

GGuillaumeLakano Hi! In the Known Issues for Local Dev, the documentation explain the producer an...

GuillaumeLakano•10/19/23, 8:28 AM

From the documentation: « Each queue can only have one consumer Worker connected to it. If you try to connect multiple consumers to the same queue, you will encounter an error when attempting to publish that Worker. »
So, if we have different workers that are producers to the same queue, how can we make only one consumer in local dev ?

CCharlie Hey @GuillaumeLakano great question. No need to copy the consumer to each worker...

GuillaumeLakano•10/20/23, 4:44 PM

Thank you for the answer, I'll try, but from what I've understood from documentation, this wasn't possible if your producers aren't in the same worker.
We have around ~20 workers that are producers ( to push in the queue an email notification to send ), but we only want one consumer in a separate worker.

iustin.pericica•10/20/23, 5:19 PM

Iiustin.pericica Hi there, We have a project based on cloudflare workers & queues. We currently h...

Matt Silverlock•10/21/23, 3:17 PM

What is your configuration? Do you have a high rate of errors? Concurrency backs off (per the docs) when errors increase.

Old Man Maker•10/22/23, 7:57 AM

I've got a dead basic consumer and I am getting this message:

Unexpected fields found in queues.consumers[0] field: "services"

Unexpected fields found in queues.consumers[0] field: "services"

I am unsure what this is refering to?

Old Man Maker•10/22/23, 7:58 AM

This is the consumer:

async queue(batch: MessageBatch<TransactionModel>, env: Bindings): Promise<void> {
        
        try {
        
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${JSON.stringify(messages)}`);


        } catch (error) {
            console.log('Error=>',error);
        }
      },

async queue(batch: MessageBatch<TransactionModel>, env: Bindings): Promise<void> {
        
        try {
        
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${JSON.stringify(messages)}`);


        } catch (error) {
            console.log('Error=>',error);
        }
      },

kian•10/22/23, 9:19 AM

It's in your Wrangler.toml

Kkian It's in your Wrangler.toml

Old Man Maker•10/22/23, 11:41 AM

Yeah - not sure if it should be classified as a bug though? The definition of a service worker anywhere lower in the toml appears to cause these types of error messages. Because mine was below the queue consumer, it looked like it was an error attached to the queue def

Moving it to the top sorted it

kian•10/22/23, 11:45 AM

It’s not a bug, just how TOML works

kian•10/22/23, 11:45 AM

You can change to using [[services]] with keys below it instead of services = [{…}]

I discovered a bug related to the queue `contentType` option [1] and dead letter queues [2]. When `c

Similar Threads