Am I doing something wrong? - Cloudflare Developers

© 2026 Hedgehog Software, LLC

More

Communities Docs About Terms Privacy

maguayo

M

maguayoOP•11/28/22, 1:22 PM

It doesnt work with wrangler either

maguayo

M

maguayoOP•11/28/22, 1:22 PM

{
  "result": null,
  "success": false,
  "errors": [
    {
      "code": 11001,
      "message": "workers.api.error.queue_handler_missing"
    }
  ],
  "messages": []
}

{
  "result": null,
  "success": false,
  "errors": [
    {
      "code": 11001,
      "message": "workers.api.error.queue_handler_missing"
    }
  ],
  "messages": []
}

maguayo

Mmaguayo Am I doing something wrong?

Chaika

C

Chaika•11/28/22, 2:03 PM

just because it's worth double checking, Do you actually have a queue handler in your code? i.e a queue function for module workers, or an event listener added to queue for service workers?
https://developers.cloudflare.com/queues/javascript-apis/#consumer

Chaika

C

Chaika•11/28/22, 2:04 PM

That error is a bit confusing at first because it kind of looks like an internal error (or at least it did to me), but it's just saying it thinks you're missing a literal queue handler in your code

maguayo

M

maguayoOP•11/28/22, 2:04 PM

Yes

Screenshot_2022-11-28_at_15.04.30.png

maguayo

M

maguayoOP•11/28/22, 2:05 PM

I thought it had to do with the "name" or something not from the code itself

Jacob

J

Jacob•11/28/22, 2:05 PM

What does your wrangler.toml look like?

Jacob

JJacob What does your wrangler.toml look like?

maguayo

M

maguayoOP•11/28/22, 2:06 PM

name = "xxx-edge"
workers_dev = true
compatibility_date = "2022-11-27"
main = "./src/index.ts"
account_id= "xxxx"

d1_databases = [
    { binding = "DB", database_name = "xxxx", database_id = "xxxx", preview_database_id = "xxxx" },
]

[[queues.producers]]
  queue = "main-queue"
  binding = "MAIN_QUEUE"

[[queues.consumers]]
  queue = "main-queue"
  max_batch_size = 1
  max_batch_timeout = 30

name = "xxx-edge"
workers_dev = true
compatibility_date = "2022-11-27"
main = "./src/index.ts"
account_id= "xxxx"

d1_databases = [
    { binding = "DB", database_name = "xxxx", database_id = "xxxx", preview_database_id = "xxxx" },
]

[[queues.producers]]
  queue = "main-queue"
  binding = "MAIN_QUEUE"

[[queues.consumers]]
  queue = "main-queue"
  max_batch_size = 1
  max_batch_timeout = 30

maguayo

M

maguayoOP•11/28/22, 2:07 PM

But since the wrangler.toml doesnt create the binding I am creaing it manually, but it doesnt work either

Jacob

J

Jacob•11/28/22, 2:08 PM

I feel like I remember someone saying we can't have a producer and consumer in the same Worker just yet, but I could be wrong

Chaika

C

Chaika•11/28/22, 2:11 PM

The example has that: https://developers.cloudflare.com/queues/examples/send-errors-to-r2/ and I'm doing that as well in one of my workers, unless I'm confusing what you're saying

Jacob

J

Jacob•11/28/22, 2:12 PM

Okay nvm, I must be miss-remembering

Jacob

J

Jacob•11/28/22, 2:13 PM

ah yep mine has the producer/consumer in the same worker

maguayo

Mmaguayo ``` name = "xxx-edge" workers_dev = true compatibility_date = "2022-11-27" main...

Jacob

J

Jacob•11/28/22, 2:14 PM

Can you try adding max_retriesmax_retries to your consumer? Also try increasing max_batch_sizemax_batch_size? Just to confirm there's not a bug with those values

Jacob

JJacob Can you try adding `max_retries` to your consumer? Also try increasing `max_batc...

maguayo

M

maguayoOP•11/28/22, 2:16 PM

Tried that, and not working either. But what worries me, is that even the GUI:
https://dash.cloudflare.com/xxxxxxxx/workers/services/view/xxxxxx/production/triggers

is not working

maguayo

M

maguayoOP•11/28/22, 2:17 PM

I am getting this error too in the GUI. So maybe is there something wrong?

Screenshot_2022-11-28_at_15.16.56.png

maguayo

Mmaguayo I am getting this error too in the GUI. So maybe is there something wrong?

Jacob

J

Jacob•11/28/22, 2:18 PM

Have you published the Worker that contains the exported queuequeue handler? Or has it always failed to publish? If so, the GUI would (correctly) think that you don't have a handler

Jacob

J

Jacob•11/28/22, 2:19 PM

To test, remove the Consumer from your wrangler.toml, publish the worker, and then try the GUI

maguayo

M

maguayoOP•11/28/22, 2:19 PM

The publish is made. The code is live and the fetch is working

maguayo

M

maguayoOP•11/28/22, 2:19 PM

Ok, I will try that

maguayo

M

maguayoOP•11/28/22, 2:25 PM

Still not working, even removing the consumers and queues from wrnagler. The dashboard doesn't work, but the code is there to handle the queue

maguayo

M

maguayoOP•11/28/22, 2:26 PM

Or maybe I am doing the handle wrong:

export default {
  fetch: app.fetch,
  scheduled: async (event, env, ctx) => {
    ctx.waitUntil(cronTrigger(event, env));
  },
  queue: async (batch: MessageBatch<Error>, env: Environment) => {
    fetch("https://api.pushover.net/1/messages.json", {
      method: "POST",
      headers: {
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: new URLSearchParams({
        token: "xxx",
        user: "xxx",
        title: "Hono",
        message: "Example of cron and queue",
      }),
    });
  },
};

export default {
  fetch: app.fetch,
  scheduled: async (event, env, ctx) => {
    ctx.waitUntil(cronTrigger(event, env));
  },
  queue: async (batch: MessageBatch<Error>, env: Environment) => {
    fetch("https://api.pushover.net/1/messages.json", {
      method: "POST",
      headers: {
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: new URLSearchParams({
        token: "xxx",
        user: "xxx",
        title: "Hono",
        message: "Example of cron and queue",
      }),
    });
  },
};

Jacob

J

Jacob•11/28/22, 2:27 PM

that fetch should probably be awaited (that likely isn't causing this issue, just an aside)

Rob

RRob I would also suggest that sleeping in your consumer Worker is an anti-pattern.

Kai

K

Kai•11/28/22, 2:29 PM

I would agree, but I'm not sure what the alternative is. Right now I'm thinking to just use a batchSize of 1 so thundering herd would be no problem. But I'm really not sure what to do over waiting when I've hit the rate limit. There seems to be no way of blocking the queue otherwise. (Of course I'd prefer to do something like "redeliver this message & don't deliver anything before x", or even just reading from the queue when an alarm is triggered).

Jacob

J

Jacob•11/28/22, 2:32 PM

I implemented a throttled queue to manage rate limits for a project in the past where I had to limit to 3rps to the Notion API: https://github.com/jahands/msc-utils/blob/main/src/notion.ts
Could probably use this when processing batches

Kai

K

Kai•11/28/22, 2:33 PM

So currently you could set some sort of semaphore using a DO.

Can you explain how this would work? I can sync the rate limit via one or multiple DOs but this is very complicated to do (especially since I have multiple "layered" rate limits) & DOs aren't really atomic iirc.

Jacob

JJacob I implemented a throttled queue to manage rate limits for a project in the past ...

Jacob

J

Jacob•11/28/22, 2:35 PM

I could see this working pretty well in a Queue consumer, and it feels like a pretty common pattern that you might need to rate-limit processing of items in a batch
It's used like this:

await queue.add(myAsyncFunction)
await queue.clear() // Execute functions in a rate-limited fassion

await queue.add(myAsyncFunction)
await queue.clear() // Execute functions in a rate-limited fassion

p-throttle and p-queue on npm are really helpful packages for stuff like this

Rob

RRob At GA we envision something *like* (not the actual API surface - still under con...

Kai

K

Kai•11/28/22, 4:08 PM

Hmm, that would be nice. But wouldn't this be more expensive to the user then just handling the timeout themselves? From what I understand each retry incurs one operation, but consumers are executed as unbound workers, so sleeping for 20s doesn't incurr any charges.

SananGuliyev

SSananGuliyev I am on enterprise plan. But unfortunately I have no contact to account team.

Unsmart

U

Unsmart•11/28/22, 6:04 PM

Account team meaning your CSM. Are you the owner of the account? If not I would reach out to whoever at your company is to find out who the CSM is to contact. Or possibly create a support ticket to see if they can help you get access?

Kai

K

Kai•11/28/22, 7:00 PM

Yes, but I'd use retries to bounce off above-ratelimit requests in this scenario, right? It's not that unlikely to have a lot of requests above that rate limit especially during peak hours

Kai

K

Kai•11/28/22, 7:18 PM

that's actually very fair. Specifying larger durations may be viable. Not sure what batch size would do here though? Is there something I'm missing about how batches work?

name = "xxx-edge"
workers_dev = true
compatibility_date = "2022-11-27"
main = "./src/index.ts"
account_id= "xxxx"

d1_databases = [
    { binding = "DB", database_name = "xxxx", database_id = "xxxx", preview_database_id = "xxxx" },
]

[[queues.producers]]
  queue = "main-queue"
  binding = "MAIN_QUEUE"

[[queues.consumers]]
  queue = "main-queue"
  max_batch_size = 1
  max_batch_timeout = 30

name = "xxx-edge"
workers_dev = true
compatibility_date = "2022-11-27"
main = "./src/index.ts"
account_id= "xxxx"

d1_databases = [
    { binding = "DB", database_name = "xxxx", database_id = "xxxx", preview_database_id = "xxxx" },
]

[[queues.producers]]
  queue = "main-queue"
  binding = "MAIN_QUEUE"

[[queues.consumers]]
  queue = "main-queue"
  max_batch_size = 1
  max_batch_timeout = 30

export default {
  fetch: app.fetch,
  scheduled: async (event, env, ctx) => {
    ctx.waitUntil(cronTrigger(event, env));
  },
  queue: async (batch: MessageBatch<Error>, env: Environment) => {
    fetch("https://api.pushover.net/1/messages.json", {
      method: "POST",
      headers: {
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: new URLSearchParams({
        token: "xxx",
        user: "xxx",
        title: "Hono",
        message: "Example of cron and queue",
      }),
    });
  },
};

export default {
  fetch: app.fetch,
  scheduled: async (event, env, ctx) => {
    ctx.waitUntil(cronTrigger(event, env));
  },
  queue: async (batch: MessageBatch<Error>, env: Environment) => {
    fetch("https://api.pushover.net/1/messages.json", {
      method: "POST",
      headers: {
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: new URLSearchParams({
        token: "xxx",
        user: "xxx",
        title: "Hono",
        message: "Example of cron and queue",
      }),
    });
  },
};