- Cloudflare Developers

MMatt Silverlock Click to see attachment

B

Bram•1/7/24, 3:58 PM

Hey, I saw that indeed. What I would like to do is something along the lines of:

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

if (msg.queue === env.SOME_QUEUE_BINDING.name) ...

B

Bram•1/7/24, 4:00 PM

Especially considering that every environment can have a different queue name bound. I can work around it by using a naming convention (e.g.

staging.some-queue.v1

staging.some-queue.v1

staging.some-queue.v1

staging.some-queue.v1 ) and so on, but was wondering if we could derive the queue name from the binding directly

A

Ashley•1/9/24, 10:54 PM

Messages delivered to a DLQ without an active consumer will persist for four (4) days before being deleted from the queue.

Is this the same for regular queues (e.g. not DLQs, the main queues themselves), where if there is no active consumer, messages will be deleted after 4 days? I couldn't see anything in the docs clarifying this, so was curious

AAshley > Messages delivered to a DLQ without an active consumer will persist for four (...

M

Matt SilverlockOP•1/10/24, 1:24 AM

https://developers.cloudflare.com/queues/platform/limits/

Limits · Cloudflare Queues

Many of these limits will increase during Queues’ public beta period. Follow our changelog or join the #queues-beta channel in our Developer Discord …

M

Matt SilverlockOP•1/10/24, 1:24 AM

See “maximum retention period”

A

Ashley•1/10/24, 9:23 AM

Thank you!

A

Abdallah Abis•1/12/24, 7:05 PM

Hey folks,
I'm working on a high-throughput service where we need to process around 1000 requests per second through a Cloudflare Worker. The goal is to rapidly queue these requests and consume them to update a database efficiently.

I have set up a producer worker to handle incoming requests. This worker assigns each request to one of 10 queues (chosen randomly) to manage the load.
The reason for multiple queues is to circumvent the limit of 20 concurrent consumers per queue. By having 10 queues, I aim to effectively handle 200 concurrent consumers (20 per queue).
The consumer worker is tasked with processing messages from each queue and updating the database.

Despite this setup, there's a noticeable lag in processing messages to the database. When I check the queue dashboard, I find that some queues still have messages pending even several minutes after they were posted. This delay is a critical issue for us, as our operation relies on rapid data processing and updates.

Any thoughts?

Thanks in advance.

U

Unsmart•1/12/24, 7:07 PM

Queues is a beta service so I would recommend using another option for production critical services.

A

Abdallah Abis•1/12/24, 7:24 PM

is the concurrency set per queue or per consumer. like does even my thought make sense?

D

divby0•1/15/24, 12:50 PM

https://developers.cloudflare.com/queues/reference/local-development/#known-issues

This known issue is still holding me back so much, it's not even funny. I am already having a worker next to my cf-pages app so I can consume from a queue to then call an api route of my pages app (because only workers can consume, pages apps can't). Now this isn't even really locally usable since we can't queue and consume from different pages/worker apps. How am I supposed to just use queues with pages apps?

Ddivby0 https://developers.cloudflare.com/queues/learning/local-development/#known-issue...

D

divby0•1/15/24, 12:56 PM

3 months later, we still basically can't even use queues if we have a pages app (since they can't consume and local dev doesn't work like that). Is that just something cloudflare doesn't wants us to use?

J

James•1/15/24, 5:29 PM

There's a lot of ongoing work with Pages + Workers convergence, but it's... slow and there's not a lot of well communicated expectations. Honestly, my stack recently when using Cloudflare has been Pages for the frontend, but offloading all the actual "work" to Workers, so I can use the rest of the ecosystem much easier, use a

wrangler.toml

wrangler.toml

, etc.

A

Al Hill•1/16/24, 10:01 AM

Sending a message from a worker to a queue count as a subrequest?

F

forslunds•1/16/24, 3:26 PM

I feel you

It looks like a long wait until something better than your solution. (Myself is sticking with GCT until its ”fixed”. Works great in dev also)

JJames There's a lot of ongoing work with Pages + Workers convergence, but it's... slow...

D

divby0•1/17/24, 11:39 AM

I can't easily do that because I use remix and don't want to split into a real frontend + backend.

As a workaround for everyone that's interested:
I have a queueService that basically serves as an adapter in my pages app. The producing into the queue fortunately is straightforward and for consuming I've got a worker that just calls api routes from my pages app and those routes use types I have defined in the respective producers (so I always only consume whatever I produced before in the same app).

As for the local dev, the queue producing in my queueService detects if I am on prod and uses normal queue and on local dev it doesn't push into a queue but instead makes an http request to a specific route on the worker that then proxies this back to the pages app (which isn't ideal cause things don't queue up, they simply are "let through". But it's well enough for local dev in my use case)

V

Vikash•1/18/24, 10:14 AM

My queue messages are duplicate processed. Is there any suggestion how to prevent that?

This is my consumer code -

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

 async queue(batch: MessageBatch<Message>, env: Env): Promise<void> {
    batch.ackAll();
    for (const message of batch.messages) {
      const task = message.body as unknown as Task;
      // d1.insert(`Running task ${task.id}`); 

      // Heavy processing here may take upto 1 minute per batch
   }
 }

Ddivby0 3 months later, we still basically can't even use queues if we have a pages app ...

M

Matt SilverlockOP•1/18/24, 11:30 AM

To be clear, Queues is still in beta, and we’re working on major asks around increased throughput, message delay controls, and addressing some bugs that affect consumer processing (per threads here). Hold tight.

MMatt Silverlock To be clear, Queues is still in beta, and we’re working on major asks around inc...

V

Vikash•1/18/24, 11:48 AM

Can I also add few more suggestion ?

Get messages -
```
queue.getMessages()
```
queue.getMessages()
```
queue.getMessages()
```
queue.getMessages()
to return an array[] will helps us to find if a message already exists with retry
Get count -
```
queue.getCount()
```
queue.getCount()
```
queue.getCount()
```
queue.getCount()
to return the number of messages currently in queue to wait or move some tasks to other services.
Is retried -
```
message.retryCount()
```
message.retryCount()
```
message.retryCount()
```
message.retryCount()
to get if a message is already retried > 3 times to take action

thanks for such an amazing product

VVikash Can I also add few more suggestion ? - Get messages - `queue.getMessages()` to...

M

Matt SilverlockOP•1/18/24, 12:13 PM

How would the first work? What is it meant to return? A queue can have tens of millions of messages.

Z

zegevlier•1/18/24, 12:14 PM

Also: If you want to find if a message is already in the queue, wouldn't a method for that make more sense than one to get all messages?

MMatt Silverlock How would the first work? What is it meant to return? A queue can have tens of m...

V

Vikash•1/18/24, 12:23 PM

A limit, offset can be used to fetch 10 or 100. I had a requirement to remove certain messages, because customer started a job and then deleted their account. So I need to find associated messages and cancel them

VVikash A limit, offset can be used to fetch 10 or 100. I had a requirement to remove ce...

V

Vikash•1/18/24, 12:25 PM

The list option already available on queue dashboard, I think you can just expose the same in worker API with some filter capabilities

VVikash The list option already available on queue dashboard, I think you can just expos...

M

Matt SilverlockOP•1/18/24, 12:33 PM

The list is more of a “peek” for the most recent messages.

It is not for finding any message in the backlog/queue - that’s a non-trivial problem. If you had to paginate through thousands or millions of messages it would not be particularly efficient (and you would pay for reads).

If you have “stale” messages it might be better to do a KV read or D1 read in your consumer and delete (ack) messages that are no longer relevant vs trying to proactively find it in the queue.

MMatt Silverlock The list is more of a “peek” for the most recent messages. It is not for findin...

V

Vikash•1/18/24, 12:38 PM

Make sense, so any function to acknowledge message by id, if we store the {message_id, org_id} in D1 ? Something like ackMessage(Id)’

VVikash Make sense, so any function to acknowledge message by id, if we store the {messa...

M

Matt SilverlockOP•1/18/24, 12:41 PM

Today, I would put the customer ID in my messages and do a D1 query to see if that customer is_deleted or not in my consumer when I read each message.

MMatt Silverlock Today, I would put the customer ID in my messages and do a D1 query to see if th...

V

Vikash•1/18/24, 12:45 PM

Yes, that’s our approach for now. But I felt consumer is receiving these messages unnecessary. So should be a way to acknowledge all by id/filter/namespace

VVikash Yes, that’s our approach for now. But I felt consumer is receiving these message...

M

Matt SilverlockOP•1/18/24, 1:16 PM

Note that if you had

ackMessages(Array<ids>)

ackMessages(Array<ids>)

ackMessages(Array<ids>)

ackMessages(Array<ids>) you'd still have to store those IDs / run that operation somewhere, and the costs would be the same (read, delete) as consuming them. This isn't to say there isn't a way to make this more convenient, but it won't land this quarter

R

Rabid•1/18/24, 6:10 PM

Whenever I add a queue binding to my wrangler.toml, or via the GUI, my cron worker no longer functions

R

Rabid•1/18/24, 6:10 PM

Usually I can go to "quick edit" > "Schedule > "Trigger Scheduled Event"

R

Rabid•1/18/24, 6:11 PM

When I have a queue binding set, this just shows a loader and does nothing, as soon as I remove the binding it starts working again

A

ajgeiss0702•1/18/24, 10:14 PM

Does the per-queue message throughput limit not apply to batch queueing? I'm generating a few thousand messages to test it, and ran them all at once (adding them with batches of 100 each) expecting an error, but they appeared to all go through?

A

ajgeiss0702•1/18/24, 10:15 PM

please ping me if you reply

T

Tuan•1/23/24, 1:33 AM

Hi

S

Supreme Commander ⚡•1/23/24, 2:13 PM

Hi, is there an appropriate channel to ask about queues functionalities besides this one?

S

switz•1/23/24, 7:20 PM

any updates on http pull + delay?

A

Ashley•1/24/24, 11:13 AM

I don't know if this is expected or not, but if you publish the same message content to a queue, should it produce another message? (I assume it should).

To simply explain, I have this line:

await env.LIVE_GAME_PRODUCER.send('check for game');

await env.LIVE_GAME_PRODUCER.send('check for game');

await env.LIVE_GAME_PRODUCER.send('check for game');

await env.LIVE_GAME_PRODUCER.send('check for game');

For reasons that aren't really important, this message is being produced every 5 minutes via a Worker cron. The first time the cron triggered, it worked, I saw a message in the dashboard, and could see logs of it being handled. Since that first message, even though the cron is triggering, no messages are reaching the queue. Is there some sort of deduplication happening? (and if so is this documented? As it seems a little odd, so hopefully I'm missing something)

A

Ashley•1/24/24, 11:40 AM

Well, actually, it's weirder. The backlog wasn't showing any, so I thought nothing was happening, but now it's refreshed so I guess messages were published. Then all of a sudden, 20+ mins after they'd been sitting in the queue, they all got consumed. I had

max_batch_size

max_batch_size

max_batch_size

max_batch_size set to 1, so not sure if that had an impact, I don't see why it would... all quite strange.

AAshley I don't know if this is expected or not, but if you publish the same message con...

M

Matt SilverlockOP•1/24/24, 12:53 PM

Yes, Queues does not (by design) deduplicate. That’s often the wrong assumption to make on behalf of users

MMatt Silverlock Yes, Queues does not (by design) deduplicate. That’s often the wrong assumption ...

A

Ashley•1/24/24, 1:27 PM

Yeah, I wouldn’t expect it too - would there be a reason for the above, where it built a backlog over 20 mins with 4 messages in the queue, but didn’t consume any? (Based on docs it would have hit the max batch or the timeout way before that)

A

Ashley•1/24/24, 2:19 PM

The metrics in the dashboard are a little slow to update, but it now shows the problem: look at the consumer delay

A

Aster•1/24/24, 3:00 PM

Hey, I am trying to use the graphQL API, but without success

I get an empty response:

{
    "data": {
        "viewer": {
            "accounts": [
                {
                    "queueMessageOperationsAdaptiveGroups": []
                }
            ]
        }
    },
    "errors": null
}

{
    "data": {
        "viewer": {
            "accounts": [
                {
                    "queueMessageOperationsAdaptiveGroups": []
                }
            ]
        }
    },
    "errors": null
}

{
    "data": {
        "viewer": {
            "accounts": [
                {
                    "queueMessageOperationsAdaptiveGroups": []
                }
            ]
        }
    },
    "errors": null
}

{
    "data": {
        "viewer": {
            "accounts": [
                {
                    "queueMessageOperationsAdaptiveGroups": []
                }
            ]
        }
    },
    "errors": null
}

And I used the query

QueueMessageOperationsByMinute

QueueMessageOperationsByMinute

QueueMessageOperationsByMinute

QueueMessageOperationsByMinute from the example (https://developers.cloudflare.com/queues/reference/metrics/), with these variables:

{
    "accountTag": "xxx",
    "queueId": "dev-event-batch-queue",
    "datetimeStart": "2024-01-24T00:00:00Z",
    "datetimeEnd": "2024-01-24T23:59:00Z"
}

{
    "accountTag": "xxx",
    "queueId": "dev-event-batch-queue",
    "datetimeStart": "2024-01-24T00:00:00Z",
    "datetimeEnd": "2024-01-24T23:59:00Z"
}

{
    "accountTag": "xxx",
    "queueId": "dev-event-batch-queue",
    "datetimeStart": "2024-01-24T00:00:00Z",
    "datetimeEnd": "2024-01-24T23:59:00Z"
}

{
    "accountTag": "xxx",
    "queueId": "dev-event-batch-queue",
    "datetimeStart": "2024-01-24T00:00:00Z",
    "datetimeEnd": "2024-01-24T23:59:00Z"
}

Annnd I do get metrics as you can see on the dashboard screenshot (I am sending requests with a curl loop to generate metrics in dev/local env)
Any idea?

AAshley Yeah, I wouldn’t expect it too - would there be a reason for the above, where it...

M

Matt SilverlockOP•1/24/24, 4:29 PM

We think we have a bug here (being a beta) when some messages take a long time to process -> cause delays in subsequent delivery. Working on it.

AAster Hey, I am trying to use the graphQL API, but without success 🤔 I get an empty ...

M

Matt SilverlockOP•1/24/24, 4:30 PM

Have you looked at the queries in https://developers.cloudflare.com/queues/reference/metrics/#get-average-queue-backlog-over-time-period

?

Metrics · Cloudflare Queues

You can view the metrics for a Queue on your account via the Cloudflare dashboard . Navigate to Workers > Queues > your Queue and under the Metrics …

MMatt Silverlock Have you looked at the queries in https://developers.cloudflare.com/queues/refer...

A

Aster•1/24/24, 4:31 PM

Yup that's the link I linked ^^

A

Aster•1/24/24, 4:32 PM

Both 3 queries return an empty array

is maybe the date incorrect? I used 24h tho

AAster Yup that's the link I linked ^^

M

Matt SilverlockOP•1/24/24, 4:33 PM

query {
  viewer {
    accounts(filter: { accountTag: $accountId }) {
      queueMessageOperationsAdaptiveGroups(
        limit: 10000
        filter: { date_geq: $startDate, date_leq: $endDate }
        orderBy: [date_DESC]
      ) {
        count
        sum {
          bytes
        }
        dimensions {
          queueId
          date
        }
      }
    }
  }
}

query {
  viewer {
    accounts(filter: { accountTag: $accountId }) {
      queueMessageOperationsAdaptiveGroups(
        limit: 10000
        filter: { date_geq: $startDate, date_leq: $endDate }
        orderBy: [date_DESC]
      ) {
        count
        sum {
          bytes
        }
        dimensions {
          queueId
          date
        }
      }
    }
  }
}

query {
  viewer {
    accounts(filter: { accountTag: $accountId }) {
      queueMessageOperationsAdaptiveGroups(
        limit: 10000
        filter: { date_geq: $startDate, date_leq: $endDate }
        orderBy: [date_DESC]
      ) {
        count
        sum {
          bytes
        }
        dimensions {
          queueId
          date
        }
      }
    }
  }
}

query {
  viewer {
    accounts(filter: { accountTag: $accountId }) {
      queueMessageOperationsAdaptiveGroups(
        limit: 10000
        filter: { date_geq: $startDate, date_leq: $endDate }
        orderBy: [date_DESC]
      ) {
        count
        sum {
          bytes
        }
        dimensions {
          queueId
          date
        }
      }
    }
  }
}

M

Matt SilverlockOP•1/24/24, 4:33 PM

{
  "data": {
    "viewer": {
      "accounts": [
        {
          "queueMessageOperationsAdaptiveGroups": [
            {
              "count": 2446,
              "dimensions": {
                "date": "2024-01-24",
                "queueId": "e4d7d7e594bd48bbbe49703f72e9c92a"
              },
              "sum": {
                "bytes": 264168
              }
            }
          ]
        }
      ]
    }
  },
  "errors": null
}

{
  "data": {
    "viewer": {
      "accounts": [
        {
          "queueMessageOperationsAdaptiveGroups": [
            {
              "count": 2446,
              "dimensions": {
                "date": "2024-01-24",
                "queueId": "e4d7d7e594bd48bbbe49703f72e9c92a"
              },
              "sum": {
                "bytes": 264168
              }
            }
          ]
        }
      ]
    }
  },
  "errors": null
}

{
  "data": {
    "viewer": {
      "accounts": [
        {
          "queueMessageOperationsAdaptiveGroups": [
            {
              "count": 2446,
              "dimensions": {
                "date": "2024-01-24",
                "queueId": "e4d7d7e594bd48bbbe49703f72e9c92a"
              },
              "sum": {
                "bytes": 264168
              }
            }
          ]
        }
      ]
    }
  },
  "errors": null
}

{
  "data": {
    "viewer": {
      "accounts": [
        {
          "queueMessageOperationsAdaptiveGroups": [
            {
              "count": 2446,
              "dimensions": {
                "date": "2024-01-24",
                "queueId": "e4d7d7e594bd48bbbe49703f72e9c92a"
              },
              "sum": {
                "bytes": 264168
              }
            }
          ]
        }
      ]
    }
  },
  "errors": null
}

M

Matt SilverlockOP•1/24/24, 4:33 PM

(As an example)

M

Matt SilverlockOP•1/24/24, 4:34 PM

I just wrote some messages into my test queue to generate some stats

Similar Threads