The issue is for one image upload, I then create 10 variations. I then need to no-op the consumer wo

Bboywonder350 The issue is for one image upload, I then create 10 variations. I then need to n...

M

Why not set up Event Notifications to only send you events for objects matching a certain prefix, or write to a different bucket, to avoid handling events you don’t want to process? Otherwise just

ack()

ack()

those messages and move on.

B

boywonder350OP•4/7/24, 9:56 PM

ooo totally fixxed the prefix or suffix documentation!

D

DanGamble•4/8/24, 9:42 AM

Given you can't use a pages app and a separate worker to consume a queue locally what is people's current workflows for handling this? I was thinking of deploying a queue and then just using that but I'm not sure how to have 1 binding be remote but the rest local

S

serkanozal•4/8/24, 6:28 PM

Hi everyone,

I am experimenting with Cloudflare Workers and Queues. During my experiments, I have noticed that sending a simple message from my Cloudflare Worker to a binded Cloudflare Queue takes about 200 ms in the warm invocations. Is this expected? Might it be caused by network latency if the binded queue is used from another data center or is it just Cloudflare Queue slow?

Especially comparing with AWS SQS send message latency, it seems too much to me. Because according to my experiments, it is around 10 ms for sending a message to AWS SQS queue, but it is 200 ms for Cloudflare Queue.

Original message was deleted

M

Matt Silverlock•4/8/24, 7:40 PM

It should look like this: https://developers.cloudflare.com/api/operations/queue-v2-messages-ack

Cloudflare API Documentation

Interact with Cloudflare's products and services via the Cloudflare API

Original message was deleted

M

Matt Silverlock•4/8/24, 8:07 PM

Can you confirm the path you’re hitting, just to be sure?

Original message was deleted

M

Matt Silverlock•4/8/24, 8:17 PM

The undocumented one is deprecated (and will be removed in future). Let me add a note to the docs just to be sure.

Original message was deleted

M

Matt Silverlock•4/8/24, 8:39 PM

https://github.com/cloudflare/cloudflare-docs/pull/13924

GitHub

queues: clarify undocumented endpoints by elithrar · Pull Request #...

cc @w-kuhn

J

John Spurlock•4/9/24, 11:07 PM

did queues conk out for about 5 minutes last hour? No messages delivered to

queue

queue

from around 5.23pm to 5.28pm us central

MMatt Silverlock You can apply an exponential back off on each attempt - we’ll soon be exposing t...

M

Matt Silverlock•4/10/24, 11:30 AM

@Florian the

attempts

attempts

field on messages has shipped. Docs PR (waiting on build): https://github.com/cloudflare/cloudflare-docs/pull/13648

GitHub

[QUEUES] Document how to apply a backoff alg by elithrar · Pull Req...

Documents how to apply a backoff alg when calling retry().

JJohn Spurlock did queues conk out for about 5 minutes last hour? No messages delivered to `qu...

M

Matt Silverlock•4/10/24, 1:05 PM

No widespread issues - queue ID?

Hhannes afaik queues run only in loctions where DOs run, which is like 12% of all Cloudf...

S

serkanozal•4/10/24, 5:36 PM

Thanks for the detailed answer. Is there any way to verify this (location of the queue colo)?

Original message was deleted

M

Matt Silverlock•4/11/24, 10:18 PM

You can change the # count in the UI - right above the picker. There isn’t pagination though due to the way Queues works right now.

Hhannes Hrm the queue worker should be able to fetch “/cdn-cgi/trace” and parse the colo...

S

serkanozal•4/12/24, 9:20 AM

When I request to the "<MY-DOMAIN>/cdn-cgi/trace", the colo in the response is "FRA". But as far as I understood, it is the colo of my worker, but CloudFlare queue might be served from another colo. Or does it mean that the queue itself is used from colo "FRA" too? If so, why the latency is so high (200ms)?

Hhannes Hrm the queue worker should be able to fetch “/cdn-cgi/trace” and parse the colo...

S

serkanozal•4/12/24, 9:26 AM

Also, when I call the "https://api.cloudflare.com/client/v4/accounts/<account-id>/queues/<queue-id>/messages/pull" endpoint to get message from queue, I have seen that it is responded from "FRA" colo as far as I see from the "CF-Ray" header. So does it mean that queue operation is handled at "FRA" colo, or still might it be the case that "FRA" colo gets the queue API request but it forwards to another colo (or data center) to handle the request, but since "FRA" colo is the entry point, we only see it in the "CF-Ray" header.

MMatt Silverlock No widespread issues - queue ID?

J

John Spurlock•4/12/24, 3:54 PM

just happened again from around 9.03 to 9.08am us central - same queue (

d3ec8122f30d4d69a7512199d25ce93b

d3ec8122f30d4d69a7512199d25ce93b

d3ec8122f30d4d69a7512199d25ce93b

d3ec8122f30d4d69a7512199d25ce93b)

J

John Spurlock•4/12/24, 9:47 PM

any guidance on what to do when receiving

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error

Error: Queue send failed: Internal Server Error calling

queue.send

queue.send

queue.send

queue.send ? I am retrying a few times backing off on the order of seconds, but the same error is thrown. Should these errors not be considered retryable?

V

Vikash•4/13/24, 12:50 PM

My queue is not sending any messages to the consumer, anyone from Cloudflare can help?

I already tried to disconnect > re-connect the consumer.

V

Vikash•4/13/24, 12:50 PM

Queue ID: 2610e093592e446f8fc0fe23632c527a

V

Vikash•4/13/24, 12:50 PM

Consumer details -

S

Sameer Ali•4/13/24, 9:06 PM

I am looking at REST API and trying to figureout how i can push messages to a queue outside of worker.

S

Sameer Ali•4/13/24, 9:06 PM

is there any way i can do this without creating a worker and push from there?

SSameer Ali is there any way i can do this without creating a worker and push from there?

J

John Spurlock•4/13/24, 10:34 PM

nope, a worker is the only way right now

https://developers.cloudflare.com/queues/examples/publish-to-a-queue-over-http/

S

Sameer Ali•4/13/24, 10:58 PM

Thanks John!

F

Fabian•4/16/24, 6:02 PM

Is there any plan to support grouping messages by an identifier? For example, I want to create batches of messages that (at best) all are "from" one customer ("the grouping key"). It would be only the preferred way of grouping batches, they still can consist of very different grouping keys though.
This is to send data to an upstream service and right now it would be like the following example:

Set max queue concurrency to 1 since upstream does not allow more than 100 req/s
batch_size to 100 (and in my case max wait time to 30sec because speed does not really matter)
Upstream also has a per-customer batching, not per message/action
Now the requests are worst case 100x more than needed - imagine 100 customers producing 3 messages every second. Instead of 3 batches consisting of data that can be grouped together to one request, I now have to make 100 requests, artificially "wait 1 second" and let the backlog grow. Otherwise the wait would not be required and the next batch could immediately be processed, since only 1 request of the max 100req/s was used

I don't know if this is a very specific use-case, but I imagine grouping by something could generally be applied for more things, increasing overall efficiency

F

Fabian•4/16/24, 6:09 PM

I'm considering not using queues altogether because of this and just group them in a database and run a scheduler like every hour/minute/whatever and pull data out of the DB. Even though the concept of a queue would be a perfect fit in my opinion.

N

natsumesou•4/17/24, 2:14 PM

Cloudflare Queues suddenly stopped working in my local environment.
Is anyone else experiencing the same issue?
This is causing a major hindrance to my development progress, and I'm very troubled by it.
https://github.com/cloudflare/workers-sdk/issues/5636

GitHub

🐛 BUG: local Cloudflare Queues died and did not revive. · Issue #56...

Which Cloudflare product(s) does this pertain to? Queues What version(s) of the tool(s) are you using? v3.20240404.0 What version of Node are you using? 20.10.0 What operating system and version ar...

N

Nephelophyte•4/18/24, 7:48 AM

Where do I find error logs for Queues?

Mmat apologies if this is addressed somewhere else, i wasn't able to find any referen...

J

John Spurlock•4/18/24, 5:39 PM

you may find

Promise.allSettled

Promise.allSettled

Promise.allSettled

Promise.allSettled more useful there: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/allSettled

MDN Web Docs

Promise.allSettled() - JavaScript | MDN

The Promise.allSettled() static method takes an iterable of promises as input and returns a single Promise. This returned promise fulfills when all of the input's promises settle (including when an empty iterable is passed), with an array of objects that describe the outcome of each promise.

Mmat you mean, `ctx.waitUntil(Promise.allSettled())`? i think that would work in tha...

S

serkanozal•4/18/24, 10:34 PM

As far as I understood, you are trying to achieve,

return response to client immediately, but wait worker until all the promises are completed (failed or succeeded)
report error(s) to Sentry immediately as they occurred without waiting all the promises are completed. For ex, if the
```
serviceOne
```
serviceOne
```
serviceOne
```
serviceOne
fails and
```
serviceTwo
```
serviceTwo
```
serviceTwo
```
serviceTwo
is on-going, report
```
serviceOne
```
serviceOne
```
serviceOne
```
serviceOne
's error to Sentry without waiting
```
serviceTwo
```
serviceTwo
```
serviceTwo
```
serviceTwo
to be completed

Did I understand correctly?

Mmat Yes. Or, mostly I am just pointing out that outside of a cloudflare worker the b...

M

Matt Silverlock•4/19/24, 1:52 PM

It rejects when any of the input's promises rejects, with this first rejection reason.

Per: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/all

MDN Web Docs

Promise.all() - JavaScript | MDN

The Promise.all() static method takes an iterable of promises as input and returns a single Promise. This returned promise fulfills when all of the input's promises fulfill (including when an empty iterable is passed), with an array of the fulfillment values. It rejects when any of the input's promises rejects, with this first rejection reason.

Mmat the `Promise.all()` rejects, but the underlying incomplete promises still run/co...

M

Matt Silverlock•4/19/24, 4:44 PM

What would you expect to happen? https://developers.cloudflare.com/workers/platform/limits/#how-long-can-a-subrequest-take covers the behaviors here. Would you want the promises to run to completion regardless (and consume CPU, even after the response has been sent?)

Cloudflare Docs

Limits · Cloudflare Workers docs

Cloudflare Workers plan and platform limits.

Mmat Yes. Or, mostly I am just pointing out that outside of a cloudflare worker the b...

S

serkanozal•4/19/24, 9:22 PM

Also are you sure that

Sentry.captureException(error);

Sentry.captureException(error);

Sentry.captureException(error);

Sentry.captureException(error); reports error before the request is completed. AFAIK, Sentry reports errors asynchronously, so when you call

Sentry.captureException(error);

Sentry.captureException(error);

Sentry.captureException(error);

Sentry.captureException(error); in the finally and let the worker completed, the error might be buffered by Sentry SDK, but not reported yet if I am not missing anything. So you may need to call

await Sentry.flush();

await Sentry.flush();

await Sentry.flush();

await Sentry.flush(); after then.

L

luna•4/19/24, 11:56 PM

is there a reason queues cant be created with capitals? all other bindings seem fine with them.

Lluna is there a reason queues cant be created with capitals? all other bindings seem ...

J

John Spurlock•4/20/24, 12:06 AM

I would assume the naming constraints (which are dns name constraints) leave open the future possibility of exposing queue-specific edge endpoints for sending/pulling without, for example, having to route pull requests through the general cf rest api

sort of like the r2 bucket-specific endpoints

JJohn Spurlock I would assume the naming constraints (which are dns name constraints) leave ope...

L

luna•4/20/24, 12:10 AM

gotcha

L

luna•4/20/24, 1:22 AM

would i expect to see logs from queues in my tail worker?

Mmat I am using toucan-js, I haven’t looked at their implementation at all, but it is...

S

serkanozal•4/20/24, 6:09 AM

If I understood your case correctly and not missing any point, does something like the following work for you?

async function waitForAll(promises) {
    await Promise.allSettled(
        promises
            .map(p => p.then(
                (x) => x,
                (e) => {
                    Sentry.captureException(e);
                    throw e;
                })
            )
    );
    // Not sure it is needed, it might not be needed
    await Sentry.flush();
}

async queue(batch, env, ctx) {
    try {
        // Schedule sending events asynchronously 
        const eventsPromises = [
            serviceOne.sendEvents(batch),
            serviceTwo.sendEvents(batch)
        ];

        // Wait event promises without blocking the response
        ctx.waitUntil(waitForAll(eventsPromises));
    } catch (error) {
        // Errors occurred during event sending are already reported to Sentry by "waitAll" above.
        // Here we are just reporting other errors if they are. 
        Sentry.captureException(error);
    }
}

async function waitForAll(promises) {
    await Promise.allSettled(
        promises
            .map(p => p.then(
                (x) => x,
                (e) => {
                    Sentry.captureException(e);
                    throw e;
                })
            )
    );
    // Not sure it is needed, it might not be needed
    await Sentry.flush();
}

async queue(batch, env, ctx) {
    try {
        // Schedule sending events asynchronously 
        const eventsPromises = [
            serviceOne.sendEvents(batch),
            serviceTwo.sendEvents(batch)
        ];

        // Wait event promises without blocking the response
        ctx.waitUntil(waitForAll(eventsPromises));
    } catch (error) {
        // Errors occurred during event sending are already reported to Sentry by "waitAll" above.
        // Here we are just reporting other errors if they are. 
        Sentry.captureException(error);
    }
}

async function waitForAll(promises) {
    await Promise.allSettled(
        promises
            .map(p => p.then(
                (x) => x,
                (e) => {
                    Sentry.captureException(e);
                    throw e;
                })
            )
    );
    // Not sure it is needed, it might not be needed
    await Sentry.flush();
}

async queue(batch, env, ctx) {
    try {
        // Schedule sending events asynchronously 
        const eventsPromises = [
            serviceOne.sendEvents(batch),
            serviceTwo.sendEvents(batch)
        ];

        // Wait event promises without blocking the response
        ctx.waitUntil(waitForAll(eventsPromises));
    } catch (error) {
        // Errors occurred during event sending are already reported to Sentry by "waitAll" above.
        // Here we are just reporting other errors if they are. 
        Sentry.captureException(error);
    }
}

async function waitForAll(promises) {
    await Promise.allSettled(
        promises
            .map(p => p.then(
                (x) => x,
                (e) => {
                    Sentry.captureException(e);
                    throw e;
                })
            )
    );
    // Not sure it is needed, it might not be needed
    await Sentry.flush();
}

async queue(batch, env, ctx) {
    try {
        // Schedule sending events asynchronously 
        const eventsPromises = [
            serviceOne.sendEvents(batch),
            serviceTwo.sendEvents(batch)
        ];

        // Wait event promises without blocking the response
        ctx.waitUntil(waitForAll(eventsPromises));
    } catch (error) {
        // Errors occurred during event sending are already reported to Sentry by "waitAll" above.
        // Here we are just reporting other errors if they are. 
        Sentry.captureException(error);
    }
}

The issue is for one image upload, I then create 10 variations. I then need to no-op the consumer wo

Similar Threads

Similar Threads

Similar Threads