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 AliOP•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);
    }
}

O

oof2win2•4/20/24, 2:57 PM

can i broadcast messages within a queue? say i have one producer and 1k consumers (consumers are not on CF but are pull consumers) and i want to send the message to all 1k consumers, i don't really want to have 1k queues for each consumer since that doesn't make too much sense

O

oof2win2•4/20/24, 2:59 PM

on the other hand i have 10k queues per account... so i could create a queue for every client (though i'm pretty sure that is not something i want)

O

oof2win2•4/20/24, 3:06 PM

even so, i need a way to broadcast to all 1k queues if i have a queue per consumer. does that exist in CF? checking the changelog that doesn't exist

A

avsaase•4/20/24, 3:44 PM

Broadcast messages are what pubsub is for.

O

oof2win2•4/20/24, 3:44 PM

yeah but does cf have that

O

oof2win2•4/20/24, 3:46 PM

oh its in a private beta

O

oof2win2•4/20/24, 3:46 PM

any eta on that?

O

oof2win2•4/20/24, 3:47 PM

the issue with queues is theres the subrequest limit of 50 reqs

J

John Spurlock•4/20/24, 4:27 PM

Ran across a new

.send

.send

.send

.send error today:

Error: Queue send failed: Conflict

Error: Queue send failed: Conflict

Error: Queue send failed: Conflict

Error: Queue send failed: Conflict

Is this something we should consider transient and immediately retry?

Ooof2win2 the issue with queues is theres the subrequest limit of 50 reqs

O

oof2win2•4/20/24, 5:31 PM

i mean worst case i send data to a queue to be distributed to individual queues…. it sounds horrendous as i would be making my own pubsub system essentially but it would work

A

avsaase•4/20/24, 8:26 PM

We are also waiting for pubsub. It would make a lot of things much simpler.

JJohn Spurlock Ran across a new `.send` error today: `Error: Queue send failed: Conflict` Is t...

M

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

Investigating. You shouldn't get that at all.

E

Elliot Hesp•4/22/24, 9:42 AM

Hi All,

We have a queue setup (one message per batch), where say 50 or so items are created every 60 seconds.

The handler for the queue simply sends a fetch request to an endpoint, and if the response is not a 200 we trigger an alert.

Every so often one of these fetch requests seems to hang (until the worker times out). We added in some abort controller logic to cancel the fetch after 5 seconds.

Interestingly, we actually see the request coming in to handle the healthcheck which resolves in way under the 5 seconds timeout.

Does anyone know why this might happen? I know there's a limit on concurrent fetch requests from a worker at once, but I'd have assumed each queue invocation is it's own and doesn't have this limit?

JJohn Spurlock Ran across a new `.send` error today: `Error: Queue send failed: Conflict` Is t...

M

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

Did you ever see a repeat of this?

MMatt Silverlock Did you ever see a repeat of this?

J

John Spurlock•4/24/24, 3:56 PM

Conflict

Conflict

Conflict

Conflict ~~I've only seen once~~. Still getting

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 occasionally on

.send

.send

.send

.send, which calling-worker-side retrying does not seem to fix

L

LordSilver•4/25/24, 2:33 PM

Will you always required Paid sub to use Queues or this thing is subject to change?

LLordSilver Will you always required Paid sub to use Queues or this thing is subject to chan...

M

Matt Silverlock•4/25/24, 3:41 PM

Subject to change, but need to think through failure modes, limits, and more work to enforce limits

L

LordSilver•4/25/24, 4:04 PM

i cannot even find the page where to upgrade to Paid plan...

L

LordSilver•4/25/24, 4:04 PM

clicking this brings me to the overview page

LLordSilver clicking this brings me to the overview page

M

Matt Silverlock•4/25/24, 4:08 PM

?

L

LordSilver•4/25/24, 4:08 PM

the link was bringing me to overview somehow

JJohn Spurlock any guidance on what to do when receiving `Error: Queue send failed: Internal Se...

C

carlos-alberto•4/25/24, 6:23 PM

Hey @John Spurlock , did you ever get an answer to this? We're seeing these errors more and more frequently and can't find any guidance on how to handle.

Ccarlos-alberto Hey @John Spurlock , did you ever get an answer to this? We're seeing these erro...

J

John Spurlock•4/25/24, 7:06 PM

not yet - currently i retry a few times on the caller (worker) side, and then give up and send the message to a DO for sending later

sometimes, but not always, sending to the DO also fails (with something like

Network connection lost

Network connection lost

Network connection lost

Network connection lost) - which is troubling since that is now lost payload.

I'm pretty sure this is just a function of Queues still being beta, and cf will try to make sure that

.send

.send

.send

.send does whatever it needs to do in tough situations to retry under the hood if necessary (like saving to disk when the network is down and retrying later)

JJohn Spurlock not yet - currently i retry a few times on the caller (worker) side, and then gi...

C

carlos-alberto•4/25/24, 7:15 PM

Thank you, quite furstrating but yeah, hopefully will improve when it moves to GA.

D

Dalist•4/26/24, 1:20 PM

Cloudflare UI is really buggy, there's so many bugs going on

L

luna•4/26/24, 11:40 PM

could i have some help understanding what all of the stats mean? im a little confused.

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

Similar Threads

Similar Threads

Similar Threads