what's the best practice for type safe queues? define a zod schema and check the payload, and reject

J

johtsoOP•5/16/24, 1:37 PM

also is there not a way to expicitly reject a message and have it go directly to the DLQ?

Jjohtso what's the best practice for type safe queues? define a zod schema and check the...

H

Hello, I’m Allie!•5/16/24, 1:41 PM

Assuming you control all of the producers, I would probably not bother validating it on the consumer, since consumers should only ever push valid messages

Jjohtso also is there not a way to expicitly reject a message and have it go directly to...

H

Hello, I’m Allie!•5/16/24, 1:42 PM

Ack the message then push to the DLQ manually? It isn't quite the same, but it performs the same task

HHello, I’m Allie!Ack the message then push to the DLQ manually? It isn't quite the same, but it p...

J

johtsoOP•5/16/24, 2:23 PM

yeah, thought that was probably the way to go

HHello, I’m Allie!Assuming you control all of the producers, I would probably not bother validatin...

J

johtsoOP•5/16/24, 2:23 PM

sure, the message might have been valid once upon a time, but code changes

J

johtsoOP•5/16/24, 2:24 PM

a key being renamed, and the consumer still expecting it to be there could mean an

undefined

undefined

causing all sorts of problems

Jjohtso sure, the message might have been valid once upon a time, but code changes 🙂

H

Hello, I’m Allie!•5/16/24, 2:26 PM

Maybe version it? Add a key at the top level with a version ID, then check just that

J

johtsoOP•5/16/24, 2:31 PM

when you have Zod, everything looks like a nail

Jjohtso when you have Zod, everything looks like a nail 😂

H

Hello, I’m Allie!•5/16/24, 2:33 PM

Yeah, Zod is definitely the one-hit hammer here. Just think it is a little heavy-handed to validate the entire thing when you control both sides anyway

H

Hello, I’m Allie!•5/16/24, 2:33 PM

But IDK, both work

M

Michael Wittig•5/16/24, 3:23 PM

I started to experiment with R2 Event Notifications over Queues using http pull. My feedback:
1) Default Api limit of 1200 reqs/5min is 4 req/sec seems way to low. I have 100+ EC2 instances working on a queue. I just requested a limit increase to 30000 reqs/5min. I hope it will be approved. I never hit an SQS rate limit...
2) One of the most important missing features is to extend the visibility timeout while working on a message. At the moment I can only set the visibility timeout when I poll a message. At this time, I might not know how long I need to process it.
3) Jurisdiction support is missing.
4) The R2 event should contain a property jurisdiction.
I hope that helps.

Aaviplayer Hi, what consumer about concurrency? How it works? If I send message into the qu...

J

johtsoOP•5/16/24, 10:41 PM

Is your consumer a cloudflare worker using a binding like this?

[[queues.consumers]]
queue = "my-queue"

[[queues.consumers]]
queue = "my-queue"

[[queues.consumers]]
queue = "my-queue"

[[queues.consumers]]
queue = "my-queue"

J

johtsoOP•5/16/24, 10:41 PM

if so what are your settings?

J

johtsoOP•5/16/24, 10:42 PM

you probably want to change your batch size to 1

J

johtsoOP•5/16/24, 10:42 PM

max_batch_size = 1

J

johtsoOP•5/16/24, 10:42 PM

otherwise both jobs will be given to a single consumer in a batch

J

johtsoOP•5/16/24, 10:43 PM

that should process the jobs in parallel..

J

johtsoOP•5/16/24, 10:47 PM

@aviplayer hmm.. reading this, cloudflare scales the number of consumers based on how many messages are being added to the queue, attempting to avoid the queue growing faster than it is processed. I wonder if it just behaves strangely if you have a very small number of jobs that take a long time to execute https://developers.cloudflare.com/queues/configuration/consumer-concurrency/

J

johtsoOP•5/16/24, 10:47 PM

it might only end up spinning up a single consumer

J

johtsoOP•5/16/24, 10:49 PM

@aviplayer I'm guessing your job is CPU bound rather than IO bound?

Aaviplayer Yes, I tried only with 2 messages, just send one than send second @johtso it jus...

J

johtsoOP•5/17/24, 9:13 AM

Aha, in that case, maybe better to do batches and process them in parallel in your worker using Promise.all and ack each message as and when it is processed

J

johtsoOP•5/17/24, 9:15 PM

I'd be interested to know too what the expected concurrency behaviour is with low throughput..

J

johtsoOP•5/17/24, 10:16 PM

Maybe you'd want something like a minimum concurrency setting

J

johtsoOP•5/17/24, 10:27 PM

@aviplayer another idea.. again not ideal. Have a long running durable object that uses the http pull method to get messages from the queue

P

Praefe•5/19/24, 11:21 PM

Hi team,
We're currently trying to use queues for api request fallbacks, so we've setup a queue workers project.
This worker, on fetch and queue consume, sends an API request to our server (which is covered by cloudflare), and if fails, then re-queues it (until configured retries)

Our problem is, we've got Cloudflare Zero Trust Access on our test server, and it's redirecting our queued requests in test environment to the Zero Trust Access login page. We've tried setting CF-Access-Client-Id and CF-Access-Client-Secret on the HTTP request to our server, but it appears to be getting stripped out at some point down the line (we believe that it's cloudflare doing this).

Any guidance on getting workers working together with Zero Trust Access?

P

Praefe•5/19/24, 11:23 PM

Maybe this is more of a workers question, I don't know if we try and do this anywhere else, will try and get more info on that

F

Fokke•5/20/24, 5:32 PM

I am trying to deploy the basic queue template worker (which worked last week) and am getting this error.

✘ [ERROR] A request to the Cloudflare API (/accounts/[REDACTED]/queues/[REDACTED]/consumers) failed.
workers.api.error.unknown [code: 10013]

Any idea what is going wrong here?

J

johtsoOP•5/20/24, 6:06 PM

how do I safely move a message into a "failure" queue? wont there be a chance that adding the message to the other queue succeed, but removing it from the first fails, or vice versa?

M

Maltmax•5/20/24, 7:51 PM

Hey! When looking at the docs it says that the consumer invocation duration is 15 minutes. But I seem to always run into a CPU Limit reached error. I am using the browser rendering API to screenshot a website, which works fine if I call it through my Hono endpoint. But if I, from my Hono endpoint, send the url to a queue it times out. My code is below and I would greatly appreciate any pointers as to might be wrong as I've been stuck on this for a while haha. Thanks in advance!

M

Maltmax•5/20/24, 7:51 PM

Calling this endpoint from my browser takes between 30-60 seconds and works just fine:

app.get('/manual-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    const fullPageScreenshotUrl = await takeScreenshot(c.env.MYBROWSER, c.env.SCREENSHOT_BUCKET, url); // Function that takes screenshot, uploads to R2 bucket, and returns URL
       return c.json({ fullPageScreenshotUrl });
});

app.get('/manual-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    const fullPageScreenshotUrl = await takeScreenshot(c.env.MYBROWSER, c.env.SCREENSHOT_BUCKET, url); // Function that takes screenshot, uploads to R2 bucket, and returns URL
       return c.json({ fullPageScreenshotUrl });
});

app.get('/manual-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    const fullPageScreenshotUrl = await takeScreenshot(c.env.MYBROWSER, c.env.SCREENSHOT_BUCKET, url); // Function that takes screenshot, uploads to R2 bucket, and returns URL
       return c.json({ fullPageScreenshotUrl });
});

app.get('/manual-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    const fullPageScreenshotUrl = await takeScreenshot(c.env.MYBROWSER, c.env.SCREENSHOT_BUCKET, url); // Function that takes screenshot, uploads to R2 bucket, and returns URL
       return c.json({ fullPageScreenshotUrl });
});

Calling this endpoint, always leads to the queue function timing out:

app.get('/take-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    // Add to queue
    await c.env.ScreenshotQueue.send({ url });

    return c.json({ message: 'success' });
});

export default {
    fetch: app.fetch,
    async queue(batch, env): Promise<void> {
        console.log(batch);
        const messages = JSON.stringify(batch.messages);
        console.log(`consumed from our queue: ${messages}`);
        switch (batch.queue) {
            case 'screenshot-queue':
                for (const message of batch.messages) {
                    const { url } = message.body;
                    if (!url) {
                        console.error('Missing URL');
                        return;
                    }

                    const fullPageScreenshotUrl = await takeScreenshot(env.MYBROWSER, env.SCREENSHOT_BUCKET, url);
                    console.log(fullPageScreenshotUrl)
                     message.ack();
                }
                break;
        }
    },
} satisfies ExportedHandler<Env>;

app.get('/take-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    // Add to queue
    await c.env.ScreenshotQueue.send({ url });

    return c.json({ message: 'success' });
});

export default {
    fetch: app.fetch,
    async queue(batch, env): Promise<void> {
        console.log(batch);
        const messages = JSON.stringify(batch.messages);
        console.log(`consumed from our queue: ${messages}`);
        switch (batch.queue) {
            case 'screenshot-queue':
                for (const message of batch.messages) {
                    const { url } = message.body;
                    if (!url) {
                        console.error('Missing URL');
                        return;
                    }

                    const fullPageScreenshotUrl = await takeScreenshot(env.MYBROWSER, env.SCREENSHOT_BUCKET, url);
                    console.log(fullPageScreenshotUrl)
                     message.ack();
                }
                break;
        }
    },
} satisfies ExportedHandler<Env>;

app.get('/take-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    // Add to queue
    await c.env.ScreenshotQueue.send({ url });

    return c.json({ message: 'success' });
});

export default {
    fetch: app.fetch,
    async queue(batch, env): Promise<void> {
        console.log(batch);
        const messages = JSON.stringify(batch.messages);
        console.log(`consumed from our queue: ${messages}`);
        switch (batch.queue) {
            case 'screenshot-queue':
                for (const message of batch.messages) {
                    const { url } = message.body;
                    if (!url) {
                        console.error('Missing URL');
                        return;
                    }

                    const fullPageScreenshotUrl = await takeScreenshot(env.MYBROWSER, env.SCREENSHOT_BUCKET, url);
                    console.log(fullPageScreenshotUrl)
                     message.ack();
                }
                break;
        }
    },
} satisfies ExportedHandler<Env>;

app.get('/take-screenshot', async (c) => {
    const url = c.req.query('url');
    if (!url) {
        return c.json({ message: 'Please provide a URL' });
    }
    // Add to queue
    await c.env.ScreenshotQueue.send({ url });

    return c.json({ message: 'success' });
});

export default {
    fetch: app.fetch,
    async queue(batch, env): Promise<void> {
        console.log(batch);
        const messages = JSON.stringify(batch.messages);
        console.log(`consumed from our queue: ${messages}`);
        switch (batch.queue) {
            case 'screenshot-queue':
                for (const message of batch.messages) {
                    const { url } = message.body;
                    if (!url) {
                        console.error('Missing URL');
                        return;
                    }

                    const fullPageScreenshotUrl = await takeScreenshot(env.MYBROWSER, env.SCREENSHOT_BUCKET, url);
                    console.log(fullPageScreenshotUrl)
                     message.ack();
                }
                break;
        }
    },
} satisfies ExportedHandler<Env>;

Original message was deleted

P

Praefe•5/20/24, 11:57 PM

The docs literally tell you to set cf headers???????

P

Praefe•5/21/24, 12:08 AM

So what we’re saying is that this works on our machines, but it “seems” that it’s getting stripped when setting them on a HttpClient in cloudflare workers

L

LordSilver•5/21/24, 1:22 PM

can somebody help me? i get empty payload in the queue message

L

LordSilver•5/21/24, 1:22 PM

i just copied the code from here https://developers.cloudflare.com/queues/examples/use-queues-with-durable-objects/

Cloudflare Docs

Queues - Use Queues and Durable Objects · Cloudflare Queues

The following example shows you how to write a Worker script to publish to Cloudflare Queues from within a Durable Object.

LLordSilver can somebody help me? i get empty payload in the queue message

M

Matt Silverlock•5/21/24, 2:59 PM

If you change

this.state.id

this.state.id

this.state.id

this.state.id to just "hello", does it work? What are you actually trying to do?

L

LordSilver•5/21/24, 3:07 PM

i'm just trying to get the example to work

MMatt Silverlock If you change `this.state.id` to just "hello", does it work? What are you actual...

L

LordSilver•5/21/24, 3:10 PM

it works

L

LordSilver•5/21/24, 3:11 PM

so i'm not sure why the example doesn't

LLordSilver so i'm not sure why the example doesn't

M

Matt Silverlock•5/21/24, 3:27 PM

The example (incorrectly) calls

this.state.id

this.state.id

this.state.id

this.state.id, which doesn't directly serialize to JSON. I've PR'ed the docs to fix this: https://github.com/cloudflare/cloudflare-docs/pull/14665 - it needs to be

this.state.id.toString()

this.state.id.toString()

this.state.id.toString()

this.state.id.toString()

GitHub

Update use-queues-with-durable-objects.md by elithrar · Pull Reques...

ID doesn't marshal to JSON directly. Use .toString() to convert before.

L

lukas•5/21/24, 6:01 PM

Am i able to access my env context variables (they get set in middleware) in a queue consumer? I want my queues to process some DB operations

L

lukas•5/21/24, 6:04 PM

or do I have to create a new client for my DB in each queue instance

Llukas Am i able to access my env context variables (they get set in middleware) in a q...

M

Matt Silverlock•5/21/24, 6:07 PM

A

queue()

queue()

queue()

queue() handler accepts a message, the Env, and the context per https://developers.cloudflare.com/queues/configuration/javascript-apis/#consumer

But what are you writing to env? Where would the middleware run? The env from the request (in your producer) is not the same as the one in the consumer. They are decoupled (as they should be)

Cloudflare Docs

Cloudflare Queues - JavaScript APIs · Cloudflare Queues

Cloudflare Queues is integrated with Cloudflare Workers. To send and receive messages, you must use a Worker.

what's the best practice for type safe queues? define a zod schema and check the payload, and reject

Similar Threads

Similar Threads

Similar Threads