Alternatively you can explicitly ack or retry messages: https://developers.cloudflare.com/queues/lea

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

items every

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

Bbharatsuthar07 I have attached 2 screenshots. please let me know which one is correct for batch...

Matt SilverlockOP•11/16/23, 2:28 AM

Neither is more correct. It’s certainly cheaper to batch them up on the producer side if you can.

MMatt Silverlock Neither is more correct. It’s certainly cheaper to batch them up on the producer...

bharatsuthar07•11/16/23, 4:28 AM

How can I batch at producer side, I don't know. Can you please edit this small part of code and resend please.
Suppose I want batch of 10. Then I have to slice 10 elements from my results array and then pass that to message queue and then consume it. Then try on next 10 batch at producer. Am I correct? You're saying that only right?

Bbharatsuthar07 How can I batch at producer side, I don't know. Can you please edit this small p...

Matt SilverlockOP•11/16/23, 11:29 AM

You are already batching by putting the entire result set into the queue as one message vs a loop over each result. I would suggest just trying out what you have - I would not get stuck on optimizing this right now if it works.

bharatsuthar07•11/16/23, 11:39 AM

Okay got it @Matt thanks

Matt SilverlockOP•11/16/23, 3:56 PM

Some updates for @Queues users:

We're working on landing HTTP Pull, which will allow you to poll a queue and consume it from outside of a Worker (or use a Worker in a poll-based flow vs. push-based). This will likely land just before/just after the holidays.
Queue throughput: our major milestone is about 3-5k msg/sec per queue (message size plays a role here), up from the ~400 msg/sec/queue currently

We're also working on landing controls around setting message delays - e.g.

await env.MY_QUEUE.send(msg, { deliveryDelay: 600 })

await env.MY_QUEUE.send(msg, { deliveryDelay: 600 })

on the producer side, and retry delays on the consumer side - e.g.

messages.batch.retryAfter(300)

messages.batch.retryAfter(300)

.retryAfter(seconds: number)

.retryAfter(seconds: number)

on a per-message basis. API not final.

Bringing Queues to GA late Q1/early Q2 next year alongside other scaling and performance improvements.

MMatt Silverlock Some updates for @Queues users: - We're working on landing HTTP Pull, which wil...

steyblind•11/16/23, 4:06 PM

Right now I've rigged data sync's to S3, which fires an object event to a Lambda, which pings a CF Worker to queue a processing job (which pulls from S3).

I'd MUCH RATHER use R2+Queues and avoid AWS altogether.

Is there an equivalent of an 'write event' I could use to populate a queue with messages as data is sync'd to R2 from [insert external platform here] (in this case, Airbyte, https://docs.airbyte.com/integrations/destinations/r2).

Ssteyblind Right now I've rigged data sync's to S3, which fires an object event to a Lambda...

Matt SilverlockOP•11/16/23, 5:14 PM

This is something on the R2 side, but we're also working on R2 Event Notifications - e.g. uploading to R2 via a Signed URL or the S3 API will publish an event to one of your Queues that you can consume.

Tumbzilla•11/16/23, 7:43 PM

set up a queue consumer that's not scaling as much as I'd expect. The queue currently has ~500k messages (max batch size 32), but the consumer is only firing every 5-10 minutes or so, and concurrency is topping out at 20, with long periods of 0 concurrency in between. All consumer invocations are succeeding.

How would I go about diagnosing the cause of this issue?

TTumbzilla set up a queue consumer that's not scaling as much as I'd expect. The queue curr...

Emilio•11/18/23, 1:11 AM

First thing I’d look at is how long the consumer script is taking to finish consuming a batch. For that you go look at the Worker metrics for the relevant script.

cg•11/18/23, 5:21 PM

Hey all! Is there a timeline for LogPush support in Queues?
It's quite hard to diagnose issues in Queues!

Paul Shriner•11/22/23, 4:38 PM

I cannot for the life of me figure out how to access Durable Objects from my consumer. I have been using itty-durable, but that has proven to be out of reach as I need the IRequest object. I would ideally like to use itty-durable, but at this point, just want it to work.

Unsmart•11/22/23, 4:42 PM

There is no request when using queue consumers... only a fetch has a request. You want to use the env which contains all of the bindings. The request also doesnt contain the bindings when in the fetch handler it is in the env parameter which is standard in all handlers.

Paul Shriner•11/22/23, 4:46 PM

Right, so when I try to do an env.DO.get("id"), it says that the id doesn't work. Hold on, let me get the exact error.

Unsmart•11/22/23, 4:47 PM

That is not a valid id... they look something like:

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

Unsmart•11/22/23, 4:47 PM

would also need to use

env.DO.idFromString()

env.DO.idFromString()

to transform to a DO id

Unsmart•11/22/23, 4:48 PM

or env.DO.idFromName

Paul Shriner•11/22/23, 4:48 PM

true

Paul Shriner•11/22/23, 4:48 PM

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);
console.log(user)

Paul Shriner•11/22/23, 4:49 PM

sorry, I was shorthanding, and shouldn't have

Unsmart•11/22/23, 4:50 PM

I'd imagine you probably dont want

this.

this.

should just be

env.

env.

. But whats the error if thats what you have?

Unsmart•11/22/23, 4:53 PM

Example:

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

Paul Shriner•11/22/23, 5:01 PM

When I call

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

it says that the method GetBrand cannot be found that I would like to call for data access.

when I call

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

This is the error code on the second:

Invalid Durable Object ID. Durable Object IDs must be 64 hex digits.

In both cases, the uid="GF5ZNCEJLV2X4UQ5FPMAPU6ZNRQ7OXOY3JPT6OOKUDRDEGKJX35TA2Y774"

Paul Shriner•11/22/23, 5:01 PM

Granted, in my non-consumer codes, I am using itty-durable and itty router, which it seems is not available here.

Unsmart•11/22/23, 5:11 PM

GetBrand is not a valid thing on durable objects you can only do fetch calls. An RPC like API is in the works though that would allow direct function calls.

Paul Shriner•11/22/23, 5:12 PM

Ahh, that makes sense. I can hit those methods with itty Durable, but not this way.

Paul Shriner•11/22/23, 5:16 PM

When I print out the user

console.log(JSON.stringify(user))

console.log(JSON.stringify(user))

It return an unpopulated object.

export default { async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) { const stub = env.TestDO.get(env.TestDO.idFromName('1')) const response = await stub.fetch('https://fake-host/', {}) } }

Alternatively you can explicitly ack or retry messages: https://developers.cloudflare.com/queues/lea

Similar Threads