Cloudflare Developers•3y ago

The only thing I'm thinking might be the problem is a mem leak in the csv parsing function

Ppaper___The only thing I'm thinking might be the problem is a mem leak in the csv parsin...

Matt Silverlock•11/6/23, 1:36 PM

If your decoder isn’t streaming correctly you will definitely run out of memory.

MMatt Silverlock If your decoder isn’t streaming correctly you will definitely run out of memory.

paper___OP•11/6/23, 3:26 PM

I'm using

csv-parse

csv-parse

, maybe the internal Buffer is not flushing correctly.

paper___OP•11/6/23, 3:26 PM

I'm creating a snapshot right now to check that

Nnil Hello everyone 🙂 I am having an issue with concurrency, for some reason my que...

Matt Silverlock•11/7/23, 5:51 PM

What is your account ID, and what are the settings for
max_batch_size
max_batch_size
and
max_batch_timeout
max_batch_timeout
on the queue? (I see them now, missed those).

What is your message volume during that period? 567 messages in a 24 hour period isn't a lot, and so your consumer shouldn't have to scale up. If your consumer is making calls to an external service before ack'ing that will also increase the measured delay.

Nnil Thanks Matt! This is my account ID: `34c0d8412965b33f782ad043d255542c` the queue...

Matt Silverlock•11/7/23, 5:58 PM

Can you share your message rate chart as well, for the same time period?

Nnil also regarding what you said, indeed before ack'ing the messages multiple calls ...

Matt Silverlock•11/7/23, 6:06 PM

No: batch timeout is on the producer side. It does not determine how the consumer scales. Your volume/message rate is very low, and (right now) we prefer to avoid scaling too quickly on the consumer side.

jasonkuhrt•11/7/23, 6:29 PM

Question about queues I guess I should have put here? workers-and-pages-helpAre queue producers and consumers supposed to work in local development with wrangler?

AKStriker98•11/8/23, 5:18 PM

Question about the queues beta. Given the public beta has been out for almost a year now, is there a rough timeline or progress level for how far along the public-beta is for full production? We would love to use queues for logging tasks but don’t yet feel comfortable using a beta product.

AAKStriker98 Question about the queues beta. Given the public beta has been out for almost a ...

Matt Silverlock•11/8/23, 8:19 PM

We want to get Queues into GA by end of Q1 next year. There’s architectural work to get per-queue throughput into the thousands-per-queue that’s happening behind the scenes.

lenqbit•11/9/23, 9:12 AM

Hello, what is the quickest/fastest way to purge a queue? We are in a situation where our queue has accumulated too much backlog that our consumer can not cope with it anymore.
We would like to purge the whole queue first as way to mitigate the issue before we look at finding out the underlying issue and work on a long term fix.
Is it a case of deploying a new version of our consumer that just acks the batches with say

ackAll()

ackAll()

?
Or is there an API that can be invoked from outside a worker?

lenqbit•11/9/23, 9:13 AM

I've seen that there is a way to list and ack message in the dashboard but this will not help as it limits the batch size to 100

Llenqbit Hello, what is the quickest/fastest way to purge a queue? We are in a situation ...

Matt Silverlock•11/9/23, 11:20 AM

There’s not a “purge all” method - re-deploying a consumer that acks every batch and returns immediately is the fastest way right now. Or deleting the queue/rebinding to a new queue.

rikka0w0•11/9/23, 1:01 PM

According to the JavaScript API documentation of Queue, inside the 'export default' block, the queue consumer is a separate handler(queue). Is there a way to poll a queue inside a running fetch()? Or it is possible to get notified inside a running fetch function for changes of a queue? Im looking for a way to implement communication between two running worker instances.

kian•11/9/23, 1:02 PM

Not without using something like Durable Objects where the Workers can read/write state for the other

Kkian Not without using something like Durable Objects where the Workers can read/writ...

rikka0w0•11/9/23, 2:32 PM

Will it be possible to implement such function with kv?

kian•11/9/23, 2:43 PM

It’s not ideal because KV is eventually consistent and reads are cached for up to 60 seconds, so polling doesn’t really work

kian•11/9/23, 2:44 PM

If you polled, and there was nothing, it’d still read as nothing even if your queue wrote to KV for up to 60 seconds.

Stockton•11/9/23, 4:55 PM

Are Queues right for me?

I'm looking to send many events to a queue, wait 30 seconds, and then send a single array with all accumulated events to a destination.

When I removed the batch.messages for each part so that it wouldn't process each event individually, it failed.

SStockton Are Queues right for me? I'm looking to send many events to a queue, wait 30 s...

Matt Silverlock•11/9/23, 6:11 PM

Can you expand? You control batching via your queue settings (https://developers.cloudflare.com/queues/learning/batching-retries/) - the messages in the consumer are always accessed via batch.messagesbatch.messages, whether it's a batch or 1 or 100.

Batching and Retries · Cloudflare Queues

When configuring a consumer Worker for a queue, you can also define how messages are batched as they are delivered.

MMatt Silverlock Can you expand? You control batching via your queue settings (https://developers...

Stockton•11/9/23, 6:44 PM

Thank you so much for replying. Essentially what's happened is that consumer did send the batch of events as a single array to the destination, but it thought that it failed. So it would retry (sending the same batch over and over) until it reached the retry limit.

SStockton Thank you so much for replying. Essentially what's happened is that consumer did...

Matt Silverlock•11/9/23, 6:47 PM

It would "fail" if the queuequeue handler throws an exception, otherwise returning without an error is considered success.

Matt Silverlock•11/9/23, 6:47 PM

Alternatively you can explicitly ack or retry messages: https://developers.cloudflare.com/queues/learning/batching-retries/#explicit-acknowledgement

Batching and Retries · Cloudflare Queues

When configuring a consumer Worker for a queue, you can also define how messages are batched as they are delivered.

MMatt Silverlock Alternatively you can explicitly ack or retry messages: https://developers.cloud...

Stockton•11/9/23, 6:51 PM

The batch was definitely delivered (as a single batch) because I'm not interating over batch.messages. So would I still need to explicitely ack every individual message even though I'm sending them as a single batch?

SStockton The batch was definitely delivered (as a single batch) because I'm not interatin...

Matt Silverlock•11/9/23, 7:33 PM

You can

ackAll()

ackAll()

as described there too

Stockton•11/9/23, 8:53 PM

Ahh got it. Wasn't working in dev. Thank you so much for the help.

Ashley•11/10/23, 2:14 PM

Random Q, is there a reason https://developers.cloudflare.com/workers/wrangler/commands/ doesn't have queues on it? (was looking for the commands and always go here, but can't see it on there, which is odd as other products in beta are on there

)

Commands - Wrangler · Cloudflare Workers docs

Create, develop, and deploy your Cloudflare Workers with Wrangler commands.

AAshley Random Q, is there a reason https://developers.cloudflare.com/workers/wrangler/c...

Matt Silverlock•11/10/23, 2:49 PM

No, just an omission. We'll get it fixed. It's not automatically generated.

Ashley•11/10/23, 2:49 PM

Thank you!

AAshley Thank you!

Matt Silverlock•11/10/23, 2:57 PM

https://github.com/cloudflare/cloudflare-docs/pull/11750

GitHub

Queues: Update commands.md by elithrar · Pull Request #11750 · clou...

This PR:

Adds Queues to the wrangler CLI docs.
Links to the Wrangler CLI docs from the Queues docs, identical to D1 and Hyperdrive.

Ashley•11/10/23, 2:59 PM

So speedy, thank you!

cg•11/12/23, 2:54 AM

hi, is there a way to clear a queue (that's not via dashboard)

cg•11/12/23, 2:54 AM

I have 200k messages backed up on my queue

Ccg hi, is there a way to clear a queue (that's not via dashboard)

Matt Silverlock•11/12/23, 2:54 PM

queues

Danny Navarro•11/13/23, 11:50 AM

Is there a way to delay a message with Cloudflare queues? I have AWS SQS delay queues in mind: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-delay-queues.html

Amazon SQS delay queues - Amazon Simple Queue Service

Learn about postponing the delivery of new messages for a specific number of seconds using Amazon SQS delay queues.

Danny Navarro•11/13/23, 11:51 AM

A Cron trigger that's executed only once would work for my particular case too, but not sure how to implement it cleanly

zegevlier•11/13/23, 12:02 PM

Durable object alarms would be perfect for that. You can make a DO that stores the message and sets the alarm, then once the alarm runs it cleans up the storage and adds it to the queue

Zzegevlier Durable object alarms would be perfect for that. You can make a DO that stores t...

Danny Navarro•11/13/23, 12:08 PM

It would work perfectly. Thanks

DDanny Navarro Is there a way to delay a message with Cloudflare queues? I have AWS SQS delay q...

Matt Silverlock•11/13/23, 12:41 PM

On the roadmap. Are you looking for simple delays at publish time, as part of a retry, something else?

MMatt Silverlock On the roadmap. Are you looking for simple delays at publish time, as part of a ...

Danny Navarro•11/13/23, 1:11 PM

I just want to invoke many times a cloudflare worker which will send a subrequest after 5 minutes to different targets depending on the initial parameters. For now the delay rate is fixed but in the future it may be dynamic.

Currently, we are using SQS delayed queues, but maybe with Cloudflare we don't need the queuing functionality, being able to delay may be enough. In AWS, we need some EC2 instances to be polling the queue all the time. Also, if we moved the polling to Cloudflare, we would avoid the future public IPv4 tax.

DDanny Navarro I just want to invoke many times a cloudflare worker which will send a subreques...

Matt Silverlock•11/13/23, 1:55 PM

Got it. A Durable Object alarm might be a better fit here, depending on what you're really trying to achieve (sounds like the delay is just one part of this).

MMatt Silverlock On the roadmap. Are you looking for simple delays at publish time, as part of a ...

huw•11/13/23, 9:55 PM

Just to piggyback on this, I’m using DO alarms to manage a per-API-key rate limit (one DO per topic / API key). I’d want the queue to only deliver kk items every nn seconds within a given topic. DOs satisfy this use case for now though

Hhuw Just to piggyback on this, I’m using DO alarms to manage a per-API-key rate limi...

Matt Silverlock•11/13/23, 10:02 PM

And just… backlog infinitely/drop older messages off the queue once they reach the retention limit, if your consumer cannot keep up / your (k, n) are too small?

MMatt Silverlock And just… backlog infinitely/drop older messages off the queue once they reach t...

huw•11/13/23, 10:11 PM

I thought about this, but I don’t think I have a good way of tracking when the last queue item was consumed or putting a lock on queue consumption without using a DO (maybe with D1, but that seems like a lot of overhead and still means a roundtrip to that DO’s colo), so I may as well just use a DO. If I flood my API with requests I get limited for 10 minutes; I’d rather drip-feed to it. But it’s been a while since I thought about this and I’m probably missing something

bharatsuthar07•11/15/23, 6:21 PM

I have attached 2 screenshots. please let me know which one is correct for batch processing of users array by using queue. should I add all 300 customers array simultaneously into message queue and then iterate them one by one on consumer side...
or Producer side only, I have to add all customers one by one into message queue by iterating? I have to do batch processing of customer array (elements of array are objects), so question is I should add this 300 customers array collectively inside message queue or one by one?
I attached screenshot for both versions, let me know which one is correct
my message queue name is batchKv

The only thing I'm thinking might be the problem is a mem leak in the csv parsing function

Similar Threads