I would assume the naming constraints (which are dns name constraints) leave open the future possibi

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 SpurlockOP•4/20/24, 4:27 PM

Ran across a new

.send

.send

error today:

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

occasionally on

.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 SpurlockOP•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

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.

L

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

docs have a broken link https://developers.cloudflare.com/queues/reference/how-queues-works/#producers

Lluna docs have a broken link https://developers.cloudflare.com/queues/reference/how-q...

J

John SpurlockOP•4/27/24, 1:00 AM

good catch, I believe those docs are open - I just submitted a quick fix PR: https://github.com/cloudflare/cloudflare-docs/pull/14284/files

L

luna•4/27/24, 3:12 AM

i can't find any docs on this. what is the limit of messages

.sendBatch()

.sendBatch()

.sendBatch()

.sendBatch() can send?

Lluna i can't find any docs on this. what is the limit of messages `.sendBatch()` can ...

J

John SpurlockOP•4/27/24, 3:40 AM

https://developers.cloudflare.com/queues/configuration/javascript-apis/#queue

JJohn Spurlock https://developers.cloudflare.com/queues/configuration/javascript-apis/#queue

L

luna•4/27/24, 3:44 AM

any chance we can get these docs and/or a link to docs pages like this in the jsdocs for the types? would make it a LOT easier than having to go digging into the docs for this.

P

Praefe•4/27/24, 5:15 AM

Why do queues bind via their names rather than their ids :NotLikeThis:

PPraefe Why do queues bind via their names rather than their ids :NotLikeThis:

L

luna•4/27/24, 5:26 AM

https://tenor.com/view/damn-these-good-ass-questions-good-questions-k-camp-good-ass-questions-questions-gif-12501849

Tenor

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

L

luna•4/28/24, 10:31 AM

any chance i could get some help with this?

B

Baya•4/29/24, 12:23 AM

Is there a way to clear a backlog from queues? I have huge consumer delay (6000 seconds).

O

oof2win2•4/29/24, 6:11 PM

can i send messages to a queue without having it bound to a worker? like a http api for the queue? and additionally an option to send the same message to multiple queues

Ooof2win2 can i send messages to a queue without having it bound to a worker? like a http ...

J

John SpurlockOP•4/29/24, 11:55 PM

there is no http api for sending messages other than using a worker: https://developers.cloudflare.com/queues/examples/publish-to-a-queue-over-http/

you could write a worker with multiple named queue bindings that sent the same input message to n queues using

Promise.allSettled

Promise.allSettled

inside the worker, for example

O

oof2win2•4/30/24, 6:43 AM

oh

O

oof2win2•4/30/24, 6:43 AM

i wanted to have a queue assigned on a per-user basis

O

oof2win2•4/30/24, 6:43 AM

but fair enough i guess, thanks

T

timmm•5/3/24, 2:21 AM

Is local dev possible with the producer in worker one and the consumer in worker two?

T

timmm•5/3/24, 2:21 AM

I'm using next-on-pages for worker one to produce the message and I'm not receiving it on worker two

T

timmm•5/3/24, 2:22 AM

Admittidely, this is the first time I've tried starting up two workers so maybe I'm not doing it right. But my understanding is you can start them both up and bindings should work

T

timmm•5/3/24, 2:29 AM

On nevermind! I see you can't have queues across workers atm in local dev

P

Praefe•5/3/24, 5:37 AM

When

wrangler deploy

wrangler deploy

ing multiple times quickly with queue config changes, the queue consumer config (and producer) page would crash with internal issues, and if I reaccessed it, the data hadn't been updated.

So the first time, it worked fine, but then the next times, while the command ran successfully, the web UI didn't reflect the config changes.

I would assume the naming constraints (which are dns name constraints) leave open the future possibi

Similar Threads

Similar Threads

Similar Threads