Cloudflare Developers•11mo ago

Consumer not receiving messages

Hi all, hate to jump in with questions, but saw this channel and figured I'd give it a shot. I have a queue set up on a worker as a producer/consumer combo between multiple environments. Meaning, my wrangler.toml looks something like this:

[[env.development.queues.producers]]
queue = "tasks-development"
binding = "QUEUE_TASKS"

[[env.development.queues.consumers]]
queue = "tasks-development"
max_batch_size = 10    
max_batch_timeout = 30 

[[env.staging.queues.producers]]
queue = "tasks-staging"
binding = "QUEUE_TASKS"

[[env.staging.queues.consumers]]
queue = "tasks-staging"
max_batch_size = 10    
max_batch_timeout = 30

[[env.development.queues.producers]]
queue = "tasks-development"
binding = "QUEUE_TASKS"

[[env.development.queues.consumers]]
queue = "tasks-development"
max_batch_size = 10    
max_batch_timeout = 30 

[[env.staging.queues.producers]]
queue = "tasks-staging"
binding = "QUEUE_TASKS"

[[env.staging.queues.consumers]]
queue = "tasks-staging"
max_batch_size = 10    
max_batch_timeout = 30

Where I have multiple consumer/producer pairs set, but the binding is the same. I've found that locally, the queues are working great - as soon as I send messages to the queue, my consumer is receiving them. However, when I deployed with wrangler, I am seeing the items enter the queue and get processed, but the binding seems to be failing where my consumer is never getting a MessageBatch to process. Any ideas would be greatly appreciated

11 Replies

dingir•10mo ago

@kewbish figured we can separate this out to a thread to not clog the main channel @skye_31 thanks for jumping in - this was started a few days ago. The last I left off with @kewbish, I had found that I'm getting Unknown Event when I run wrangler tail <worker-name> I haven't tried different Binding names yet for each environment that I have @skye_31 gotcha - outside of that, I have logs directly inside my consumer function that I defined that I'm not getting to The queue itself seems to still be working, in the sense that it's processing and removing items, but that logic is never ran Just using wrangler tail to see if the logs are happening Figured that if I see updates on my end, then it's working (lol) Probably - and that would cause my consumer to never be called? What's weird is that this only seems to be happening on staging/prod builds. Works perfectly on localhost That makes sense - guess I'm just confused as to how I would have different results with the same logic between environments, but this helps!

kewbish•10mo ago

Hi, just circling back here. You mentioned your per-message log is never being run - does it have some external side effects that you'd be able to see (does it call an external API for example?) Another idea you could try is adding a fetch to something like webhook.site when you loop over each message. If calls to the webhook are made, the logic is running (and we just have a bug in wrangler tail), but if they're not made then there's something else going on.

dingir•10mo ago

@kewbish apologies for the late response - I can definitely take a look into it and see what I find @kewbish side question - if I am using wrangler tail to watch the logs from my queue, would I see any data that it consumes? I'm getting Unknown Event - Ok, but no message body after it Unless I have to explicitly log it out

kewbish•10mo ago

I don't believe you'd see the data being consumed unless you log it out!

dingir•10mo ago

So interesting note - I'm running wrangler tail right now and I'm getting API logs - logs about what endpoints are being hit (since the producer/consumer are on a Hono instance) . Guess I'm not tail-ing the correct thing? My consumer/producer are named correctly though...hm

kewbish•10mo ago

I don't think you can filter tails just to the Queue events IIRC, but you should be tailing the right thing. It's probably just that your endpoints are getting hit much more frequently than your consumer is being called.

dingir•10mo ago

Yeah, that's what it is lol

kewbish•10mo ago

Have you tried adding a fetch out to a webhook site? That'd probably be the easiest way to know for certain if your consumer is running or not.

dingir•10mo ago

I can! Not sure what you mean exactly in terms of fetching to a webhook site, sorry

kewbish•10mo ago

webhook.site is just a free site that lets you generate URLs to fetch and that logs all requests coming in. You can use it to check if your consumer is running, since if the webhook shows it's been fetched then your consumer has run. You can go to webhook.site and copy 'Your unique URL', then as the very first part of your queue handler, you can try doing something like:

await fetch("[unique url here]")

await fetch("[unique url here]")

Then when you send some messages to your Queue, you can check the webhook page to see if any requests have been made. If there have been requests, your consumer's been called.

dingir•10mo ago

Oh gotcha, yeah can definitely try that Hey @kewbish looks like that works, but I did it with a barebones queue set up. Going to replicate our staging env and slowing begin adding/removing things until it doesn't work. Thanks again for all of the help and suggesstions - once I find out what it is, I'll be sure to let you guys know @kewbish turns out our queue handler wasn't await-ed 🫠 thanks again for all of the help

Gaming

Programming

Consumer not receiving messages