Can I also add few more suggestion ? - Get messages - `queue.getMessages()` to return an array[] w

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

VikashOP•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

VikashOP•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

VikashOP•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

VikashOP•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>)

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');

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

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

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

A

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

Thanks a lot! My main issue is I thought queueId was the queue name ^^

A

Aster•1/25/24, 10:31 AM

Thanks @Matt , it solved my issue btw

A

Aster•1/25/24, 10:31 AM

And for now I'll keep a Map in my code with queue Id and queue name

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

A

Ashley•1/25/24, 12:07 PM

Interestingly on this one, the queue shows 1 (out of 4) messages on the dashboard when I poll, but the graph shows 4 in the backlog... I know there's the bug so more just more info. Also, if you manage to fix it, could you let me know it's resolved so I can re-try my implementation please?

A

Ashley•1/25/24, 1:23 PM

Also, related, but... should I be hitting CPU limits when consuming? It seems... really weird I'd be hitting a CPU limit considering I'm on standard pricing, and it should be able to run presumably for... a while? (it is running for at most 10 seconds, but most should be I/O)

A

Adi•1/28/24, 4:23 AM

Hey all, I had a question around the v8 serialization that Queues does to save objects that can't be json serialized.
I'm guessing under the hood it does something like this

const v8 = require('v8');
// Serialize an object into a buffer using v8
function serializeWithV8(obj) {
    return v8.serialize(obj);
}
// Convert the Buffer to a byte array (Uint8Array)
const byteArray = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
// Now you can send the byteArray over the network

const v8 = require('v8');
// Serialize an object into a buffer using v8
function serializeWithV8(obj) {
    return v8.serialize(obj);
}
// Convert the Buffer to a byte array (Uint8Array)
const byteArray = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
// Now you can send the byteArray over the network

const v8 = require('v8');
// Serialize an object into a buffer using v8
function serializeWithV8(obj) {
    return v8.serialize(obj);
}
// Convert the Buffer to a byte array (Uint8Array)
const byteArray = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
// Now you can send the byteArray over the network

const v8 = require('v8');
// Serialize an object into a buffer using v8
function serializeWithV8(obj) {
    return v8.serialize(obj);
}
// Convert the Buffer to a byte array (Uint8Array)
const byteArray = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
// Now you can send the byteArray over the network

However, I can see that the v8 api is not available in workers under

nodejs_compat

nodejs_compat

nodejs_compat

nodejs_compat.

I was looking to making an API interface around queues for a project. Is there a chance that we can get v8 added to nodejs_compat? Is that something the team is already looking into?

K

kian•1/28/24, 4:28 AM

fwiw you can see how it works at https://github.com/cloudflare/workerd/blob/5d27f8f7f1f9b584f673d2f11c9032f5a776ec55/src/workerd/api/queue.c%2B%2B#L33-L128

Can I also add few more suggestion ? - Get messages - `queue.getMessages()` to return an array[] w

Similar Threads

Similar Threads

Similar Threads