Cloudflare Developers•2y ago

Same for me, it never worked. I am trying on local btw. Havent deployed my worker yet.

Sshreeyansb Same for me, it never worked. I am trying on local btw. Havent deployed my worke...

Matt Silverlock•5/28/24, 9:05 AM

What version of wrangler are you using? Local will be all the difference here!

MMatt Silverlock What version of wrangler are you using? Local will be all the difference here!

shreeyansbOP•5/28/24, 9:12 AM

I was on v2 of wrangler. Upgrading to v3 solved the issue. Thanks a lot.

Eengrtitus No, it never did. It always gets retried immediately

Matt Silverlock•5/28/24, 9:34 AM

Per queues - are you also using this locally?

Fokke•5/28/24, 8:16 PM

For queues is there a way to force more workers to span for consuming workload? For example, say I have 10 items in my queue, but I want them to be processed as quickly as possible. Can I force deploy 10 workers to deal with that workload?

Basically I want the opposite of max concurrency

Eengrtitus No, i am not using it locally

Matt Silverlock•5/28/24, 10:53 PM

OK, thanks. Can you confirm the wrangler version you’re using and open an issue here with a minimal reproduction? https://github.com/cloudflare/workerd/issues/new/choose

GitHub

Build software better, together

GitHub is where people build software. More than 100 million people use GitHub to discover, fork, and contribute to over 420 million projects.

GuillaumeLakano•5/29/24, 9:45 AM

Hi!

About Queues pricing, when the Queues is empty and a post-consumer Worker is triggered because of max_batch_timeout, do this cost money ? (may be a "read" from Queue, and may be CPU/Request of a consumer Worker) ?

To prevent that, could we set « set max_batch_timeout = 0 » and « max_batch_size = 1 », to not waste money please?

GGuillaumeLakano Hi! About Queues pricing, when the Queues is empty and a post-consumer Worker ...

Matt Silverlock•5/29/24, 10:40 AM

Do you mean pull consumer?

Matt Silverlock•5/29/24, 10:40 AM

A push consumer will not be invoked at all if the queue is empty, so $0.

MMatt Silverlock Do you mean pull consumer?

GuillaumeLakano•5/29/24, 10:45 AM

Not a HTTP pull consumer, the automatic push consumer, triggered when either max_batch_timeout or max_batch_size is reached
Good news if it's not invoked when the queue is empty !

Eengrtitus Hi, wrangler version is 3.57.1 But i just stumbled upon this on the queue setti...

Matt Silverlock•5/29/24, 11:21 AM

That should not affect it. In my (non) repro, my batch timeout is 30s and my retry delaySeconds is 90s. delaySeconds on an individual message will take precedence over the queue-level settings.

Sinderella•5/30/24, 1:09 PM

I am currently following this example https://github.com/cloudflare/workers-rs/blob/078d15bb60d2b05af0236c7b1370f598f5e17ada/examples/queue/src/lib.rs#L113

I am facing the following error message, similar to Messages in Queue not Processing

Queue test-queue (5 messages) - Exceeded CPU Limit @ 30/05/2024, 19:52:42

Queue test-queue (5 messages) - Exceeded CPU Limit @ 30/05/2024, 19:52:42

I am wondering how does the consumer work. Each message of mine should query a row from D1, make a HTTP request, then update the row in D1, which shouldn't take long.

Paul•5/30/24, 11:12 PM

In local development, when I have an api route that triggers a queue using hono, and the consumer simply console logs the message, for some reason it has a delay of 25 seconds before all the console logs come printing out. Any idea on why that is? I would have thought it would be instant...?

Paul•5/30/24, 11:13 PM

Code is written with hono.js

This is my producer:

app.get("/test", async c => {
    const message = `Test message. Random number: ${Math.floor(Math.random() * 100)}`;
    await c.env.WEBSITE_QUEUE.send({
        body: message,
    });

    return c.json({ success: `Queued up this message: ${message}` });
});

app.get("/test", async c => {
    const message = `Test message. Random number: ${Math.floor(Math.random() * 100)}`;
    await c.env.WEBSITE_QUEUE.send({
        body: message,
    });

    return c.json({ success: `Queued up this message: ${message}` });
});

This is my consumer that simply console.logs the message

export default {
    fetch: app.fetch,
    async queue(batch: MessageBatch<{ body: string }>, env: Environment) {
        for (const message of batch.messages) {
            console.log(
                `Consuming message: ${JSON.stringify(message.body, null, 2)}`
            );
        }
    },
};

export default {
    fetch: app.fetch,
    async queue(batch: MessageBatch<{ body: string }>, env: Environment) {
        for (const message of batch.messages) {
            console.log(
                `Consuming message: ${JSON.stringify(message.body, null, 2)}`
            );
        }
    },
};

Paul•5/30/24, 11:19 PM

Nevermind, I'm an idiot. It's the max batch timeout settings in the toml file.

lukas•5/31/24, 12:00 AM

Is the JS API the only way to retrieve a list of active queues within a worker?

At what point are queues officially completed? Obviously if something doesn't throw within the queue, it means that you can ack the message and retire it once whatever operations you have are completed (if operations complete fine, then esentially the job is done and you can mark it as completed)?

Llukas Is the JS API the only way to retrieve a list of active queues within a worker? ...

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

What do you mean by "active queues" - ? Queues that a Worker is bound (via bindings) to?

MMatt Silverlock What do you mean by "active queues" - ? Queues that a Worker is bound (via bindi...

lukas•5/31/24, 2:00 PM

Yes. I think I was thinking about this wrong. I just need to message.ack() after DB operations finish, and long poll a route that checks if the DB entries have a processedAt column entry.

Sohaib Muhammad•6/3/24, 10:43 AM

Hi, i have a question that in a producer can we get the number of messages pending in queue, this is because i have a use case in which i only need to fill queue with new messages if it is empty

Murder Chicken•6/3/24, 1:37 PM

Any plans to support queue consumer workers running locally alongside a producer worker? The current solution of putting the consumer code into the producer code to test it locally is... not ideal.

MMurder Chicken Any plans to support queue consumer workers running locally alongside a producer...

Hello, I’m Allie!•6/3/24, 1:59 PM

It should already? You mean Local Worker -> Local Queue -> Local Consumer?

Murder Chicken•6/3/24, 1:59 PM

Correct. I can't get it to work and the documentation explicitly calls out that it's not working.

MMurder Chicken Correct. I can't get it to work and the documentation explicitly calls out that...

Hello, I’m Allie!•6/3/24, 2:00 PM

Huh... I have it working for one app... Where does it say that?

Murder Chicken•6/3/24, 2:00 PM

MMurder Chicken Click to see attachment

Hello, I’m Allie!•6/3/24, 2:00 PM

Oh, separate....

Hello, I’m Allie!•6/3/24, 2:00 PM

Uh...

Hello, I’m Allie!•6/3/24, 2:00 PM

Right

Hello, I’m Allie!•6/3/24, 2:01 PM

Yeah, I guess that is just a wait until it is supported thing?

Murder Chicken•6/3/24, 2:01 PM

That's what I'm doing now but I was curious if it was a priority.

MMurder Chicken That's what I'm doing now but I was curious if it was a priority.

Hello, I’m Allie!•6/3/24, 2:02 PM

Based on the wording there, I would guess it is on the backlog, but not sure on the priority...

Murder Chicken•6/3/24, 2:07 PM

Local wrangler development with all of the features is literally one of the best features and simultaneously, one of the most confusing.

IIan I have a monorepo (managed by LernaJS) that has a cloudflare pages package and a...

johtso•6/4/24, 12:13 AM

did you figure out if using --persist-to in a monorepo is the right way to go?

Jjohtso did you figure out if using --persist-to in a monorepo is the right way to go?

Ian•6/4/24, 6:50 PM

I couldn't get it working. I wound up using a 3rd party queue service called QStash. However, this came with its own issues for doing local testing since now I need to use ngrok to create a local tunnel. You might be able to create a local tunnel using cloudflare tunnels as well.

Sohaib Muhammad•6/5/24, 11:17 AM

getting this when consumer worker get message from queue and called a service worker function "The destination execution context for this RPC was canceled while the call was still running."

kcbrewron•6/6/24, 1:26 AM

Are we still saying "soon" for general availability dates?

HHello, I’m Allie!Based on the wording there, I would guess it is on the backlog, but not sure on ...

Todd@CF•6/6/24, 3:23 PM

Yeah this is exactly correct. Running queues locally with Wrangler dev is not fully supported yet, but it is on the roadmap.

lukas•6/6/24, 10:33 PM

Huh, my consumers trigger locally?

Llukas Huh, my consumers trigger locally?

Ian•6/7/24, 5:51 AM

I think he means for Pages. From what I can tell, queues work with a workers project.

SSohaib Muhammad getting this when consumer worker get message from queue and called a service wo...

Marwan•6/8/24, 1:08 AM

غربي ؟

Rohit Verma•6/9/24, 6:52 AM

is there a way we can get the r2 file created event in the queues?

jordanfinners•6/9/24, 7:55 AM

Would it be possible to get a redrive button in the dashboard for pushing deadletter messages back onto the original queue? Please

ancs21•6/10/24, 2:08 AM

I see limit of delay message is 12 hours, can request increase this? Thanks.

ancs21•6/10/24, 2:10 AM

My team have plan move from Upstash to Queues, Upstash QStash support max delay is 7 days, also PAYG is 1 year.

ajgeiss0702•6/11/24, 6:59 PM

does the

delaySeconds

delaySeconds

option in

sendBatch

sendBatch

apply to each message in the batch? Or once for the entire batch.

E.g. if I set the delaySeconds to a minute (60) and have 5 messages in the batch, will it wait a minute and execute all 5 messages, or wait a minute and execute one, wait another minute and execute another, etc

ajgeiss0702•6/11/24, 7:03 PM

please ping me if you reply

ajgeiss0702•6/11/24, 7:14 PM

also, does local development not support the

delaySeconds

delaySeconds

option? If I try using it in local dev, I get these:

✘ [ERROR] Uncaught (in response) Error: Queue send failed: Internal Server Error

✘ [ERROR] Uncaught (in response) Error: Queue send failed: Internal Server Error

[wrangler:inf] GET /__scheduled 500 Internal Server Error (1904ms)
[wrangler-UserWorker:err] TypeError: Invalid statusText

[wrangler:inf] GET /__scheduled 500 Internal Server Error (1904ms)
[wrangler-UserWorker:err] TypeError: Invalid statusText

app.get("/test", async c => { const message = `Test message. Random number: ${Math.floor(Math.random() * 100)}`; await c.env.WEBSITE_QUEUE.send({ body: message, }); return c.json({ success: `Queued up this message: ${message}` }); });

export default { fetch: app.fetch, async queue(batch: MessageBatch<{ body: string }>, env: Environment) { for (const message of batch.messages) { console.log( `Consuming message: ${JSON.stringify(message.body, null, 2)}` ); } }, };

Same for me, it never worked. I am trying on local btw. Havent deployed my worker yet.

Similar Threads

Similar Threads

Similar Threads