There's a lot of ongoing work with Pages + Workers convergence, but it's... slow and there's not a l

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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•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 Silverlock•1/24/24, 4:33 PM

(As an example)

M

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

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

MMatt Silverlock ``` query { viewer { accounts(filter: { accountTag: $accountId }) { ...

A

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

Can you share your filters?

A

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

Note: when I remove the filter by queue, it works

And the date was in wrong format

A

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

I was using the queue name, but seems like the queue ID is different from the slug

A

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

How do you get a queue ID ? I dont see any ID from the dashboard (and is it possible to return a slug by any change in the response dimensions?

)

MMatt Silverlock We think we have a bug here (being a beta) when some messages take a long time t...

A

Ashley•1/24/24, 4:45 PM

Thank you, and no worries, not a big deal

Well, in this case it meant I exceeded browser rendering limits

But it's a beta so no sweat, thanks for looking into it!

A

Ashley•1/24/24, 4:46 PM

And what you said makes sense as processing takes ~7 seconds mostly due to time to start Pupeeter + fetch a page

AAster Can you share your filters? 🙂

M

Matt Silverlock•1/24/24, 5:00 PM

I don't have any filters: the only filter is

accountTag

accountTag

accountTag

accountTag as defined there.

AAster How do you get a queue ID ? I dont see any ID from the dashboard (and is it poss...

M

Matt Silverlock•1/24/24, 5:02 PM

Queue ID:

wrangler queues list

wrangler queues list

wrangler queues list

wrangler queues list today (we need to add it to the dash)

MMatt Silverlock I don't have any filters: the only filter is `accountTag` as defined there.

A

Aster•1/24/24, 5:02 PM

what about

startDate

startDate

startDate

startDate?

MMatt Silverlock Queue ID: `wrangler queues list` today (we need to add it to the dash)

A

Aster•1/24/24, 5:03 PM

Would it be possible to include it in the response dimensions of the graphQL?

AAster what about `startDate`?

M

Matt Silverlock•1/24/24, 5:03 PM

A

Aster•1/24/24, 5:04 PM

Sorry variables* my bad

There's a lot of ongoing work with Pages + Workers convergence, but it's... slow and there's not a l

Similar Threads

Similar Threads

Similar Threads