Hi! Is there any way of delaying a message, e.g. I want to this message to be consumed in 2 days?

TTyler Benfield @Matt do you see any potential issues with <:this:1010240499464024116> workaroun...

M

Matt Silverlock•12/1/23, 3:43 PM

I don't love the workaround (you shouldn't have to do this!) but there's no issues with it

AAlexander Blöndal Hi! Is there any way of delaying a message, e.g. I want to this message to be co...

M

Matt Silverlock•12/1/23, 3:44 PM

queues

F

Fun Miles•12/2/23, 11:21 PM

The docs (https://developers.cloudflare.com/workers/platform/pricing/) say queues worker get 15 minutes CPU times but my queue workers have gotten killed way before that (less than 5 minutes of wall time). There's a similar issue with Cron ones. Is this a known problem?

Pricing · Cloudflare Workers docs

Workers plans and pricing information.

R

RuthsPlace•12/4/23, 9:22 AM

having the same issue @Fun Miles . Any feedback?

FFun Miles The docs (https://developers.cloudflare.com/workers/platform/pricing/) say queue...

M

Matt Silverlock•12/4/23, 2:26 PM

What usage model is your Worker operating as? Also note, from the docs I linked before:

M

Matt Silverlock•12/4/23, 2:26 PM

I am being killed after 32,000ms. Is this a bug in the new billing?

32000ms = 30s (plus some leeway we provide)

F

Fun Miles•12/4/23, 2:30 PM

30s indeed but a query should go to 15 minutes, ie 900,000 ms.
My query job does D1 batched read requests with await, so most of that shouldn't count as CPU time and then some processing before some batched D1 writes.

FFun Miles 30s indeed but a query should go to 15 minutes, ie 900,000 ms. My query job does...

M

Matt Silverlock•12/4/23, 3:09 PM

Can you answer my other question? What usage model is your Worker running as? From the error you posted, your Queue consumer (I think?) is hitting the CPU limit posted above:

Queue schedule-retrieval (7 messages) - Exceeded CPU Limit @ 12/2/2023, 12:03:45 PM

Queue schedule-retrieval (7 messages) - Exceeded CPU Limit @ 12/2/2023, 12:03:45 PM

MMatt Silverlock Can you answer my other question? What usage model is your Worker running as? Fr...

F

Fun Miles•12/4/23, 3:10 PM

I don't understand what you mean by "model" then.

F

Fun Miles•12/4/23, 3:12 PM

It's the standard billing which says:

F

Fun Miles•12/4/23, 3:13 PM

As I've hit the limit both in Cron Trigger, then, after refactoring the code to divide in smaller queue works, the Queue Consumer, I am presuming there is a bug and it probably has the same root in both.

FFun Miles I don't understand what you mean by "model" then.

M

Matt Silverlock•12/4/23, 3:15 PM

https://developers.cloudflare.com/workers/platform/pricing/#how-to-switch-usage-model

your Worker will run under a usage model. You can check the dash. It will depend on when you created it.

F

Fun Miles•12/4/23, 3:17 PM

Standard model.

F

Fun Miles•12/4/23, 3:40 PM

R

RuthsPlace•12/4/23, 6:36 PM

For reference, the only way I could get more than 30 seconds was by
1) Moving from "standard" to "unbound"
2) Moving to "cron" trigger (I realize this is likely not helpful if you need queue but could work with my use case)
3) Adding this line to my wrangler.toml - workers_dev = false

R

RuthsPlace•12/4/23, 6:38 PM

You can see the success logs. The green ones are where the only change is to run under "unbound" usage model. The failed ones are where I run "standard" usage model.

R

RuthsPlace•12/4/23, 6:40 PM

Anyhow, just wanted to share in case it helps Cloudflare team diagnose issue. I don't think it's behaving as the documentation expresses it currently, unless I'm missing something

R

Ryan K•12/5/23, 12:05 AM

Hi can anyone help provide more context for this error I'm seeing when deploying workers with a queue:

Total Upload: 1810.83 KiB / gzip: 347.18 KiB
Uploaded dev-api-internal (2.44 sec)

✘ [ERROR] A request to the Cloudflare API (/accounts/2c/workers/queues/dev-queue/consumers/api-backend) failed.

  workers.api.error.queue_handler_missing [code: 11001]

Total Upload: 1810.83 KiB / gzip: 347.18 KiB
Uploaded dev-api-internal (2.44 sec)

✘ [ERROR] A request to the Cloudflare API (/accounts/2c/workers/queues/dev-queue/consumers/api-backend) failed.

  workers.api.error.queue_handler_missing [code: 11001]

Total Upload: 1810.83 KiB / gzip: 347.18 KiB
Uploaded dev-api-internal (2.44 sec)

✘ [ERROR] A request to the Cloudflare API (/accounts/2c/workers/queues/dev-queue/consumers/api-backend) failed.

  workers.api.error.queue_handler_missing [code: 11001]

Total Upload: 1810.83 KiB / gzip: 347.18 KiB
Uploaded dev-api-internal (2.44 sec)

✘ [ERROR] A request to the Cloudflare API (/accounts/2c/workers/queues/dev-queue/consumers/api-backend) failed.

  workers.api.error.queue_handler_missing [code: 11001]

K

kian•12/5/23, 12:10 AM

It's just saying that it doesn't detect a

queue

queue

function in your default export

Kkian It's just saying that it doesn't detect a `queue` function in your default expor...

R

Ryan K•12/5/23, 12:13 AM

I assumed that could be the case but I do indeed have a default export:

export default {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    // Set env variables
    Object.keys(env).forEach((variable) => {
      process.env[variable] ??= `${env[variable]}`;
    })

    return fetch(request, env, ctx)
  },
  async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
    let messages = JSON.stringify(batch.messages);
    console.log(`consumed from our queue: ${messages}`);
  },
}

export default {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    // Set env variables
    Object.keys(env).forEach((variable) => {
      process.env[variable] ??= `${env[variable]}`;
    })

    return fetch(request, env, ctx)
  },
  async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
    let messages = JSON.stringify(batch.messages);
    console.log(`consumed from our queue: ${messages}`);
  },
}

export default {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    // Set env variables
    Object.keys(env).forEach((variable) => {
      process.env[variable] ??= `${env[variable]}`;
    })

    return fetch(request, env, ctx)
  },
  async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
    let messages = JSON.stringify(batch.messages);
    console.log(`consumed from our queue: ${messages}`);
  },
}

export default {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    // Set env variables
    Object.keys(env).forEach((variable) => {
      process.env[variable] ??= `${env[variable]}`;
    })

    return fetch(request, env, ctx)
  },
  async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
    let messages = JSON.stringify(batch.messages);
    console.log(`consumed from our queue: ${messages}`);
  },
}

K

kian•12/5/23, 8:12 AM

They all should be available - have you tried them?

K

kian•12/5/23, 8:13 AM

I think that documentation aside is old, and should say within any handler context (fetch, scheduled, queue, email, etc)

K

kian•12/5/23, 4:54 PM

What version of

workers-types

workers-types

are you using? It's likely outdated

K

kian•12/5/23, 4:54 PM

The current version has this

declare interface Queue<Body = unknown> {
  send(message: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>): Promise<void>;
}

declare interface Queue<Body = unknown> {
  send(message: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>): Promise<void>;
}

declare interface Queue<Body = unknown> {
  send(message: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>): Promise<void>;
}

declare interface Queue<Body = unknown> {
  send(message: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>): Promise<void>;
}

K

kian•12/7/23, 8:17 AM

Smart Placement currently only affects

fetch

fetch

handlers iirc

M

Mikyari•12/9/23, 5:18 AM

Is there any kind of delay message like Rabbit MQ in Cloudflare Queues?

I intend to use Cloudflare Queues to define solution for my case.

User creates a new order
The system will send an email to remind user about the unpaid order after 30 minutes. If the user complete the payment within 30 minutes, the email will not be sent.

MMikyari Is there any kind of delay message like Rabbit MQ in Cloudflare Queues? I inten...

H

Hello, I’m Allie!•12/9/23, 7:33 AM

I wouldn’t think you need Queues at all. Create a DO with a 30 minute alarm. If it receives a Request before the alarm triggers, clear it

MMikyari Is there any kind of delay message like Rabbit MQ in Cloudflare Queues? I inten...

J

John Spurlock•12/9/23, 9:48 PM

not yet, but they are working on it: queues

C

cg•12/10/23, 10:09 PM

Hey @Matt , how's the investigation going, do you have any updates on timelines? Should I just assume that the fix would be in the GA in Q1/Q2 next year?

Ccg Hey @Matt , how's the investigation going, do you have any updates on timelines?...

M

Matt Silverlock•12/11/23, 12:39 AM

We’re still working on it. Non-trivial to fix. Likely within the next 2-3 weeks.

SSam We have also had this same issue with a queue that seemly just stopped sending m...

M

Matt Silverlock•12/11/23, 7:45 PM

We’re working on addressing this - it’s an unfortunate beta-type bug.

SSam We have also had this same issue with a queue that seemly just stopped sending m...

M

Matt Silverlock•12/11/23, 7:46 PM

P.S: an account ID and queue ID allow us to investigate further. I generally recommend avoiding assuming your bug == someone else’s bug

M

Matt Silverlock•12/11/23, 7:48 PM

They’re not private identifiers. You can post them here @SamP

SSam account ID is 12fd8847ed86e95913678748fab30104 and broken queue ID is seine

M

Matt Silverlock•12/11/23, 8:57 PM

Looks like the consumer is consuming fast now. We manually intervened (and have a patch going out to attempt to fix the root cause)

Hi! Is there any way of delaying a message, e.g. I want to this message to be consumed in 2 days?

Similar Threads