Am i able to access my env context variables (they get set in middleware) in a queue consumer? I wan

L

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

MMatt Silverlock A `queue()` handler accepts a message, the Env, and the context per https://deve...

L

lukasOP•5/21/24, 6:09 PM

Yes, but am i able to do

context.get('variable')

context.get('variable')

context.get('variable')

context.get('variable') within the queue? My CTX is empty within the Queue consumer

L

lukasOP•5/21/24, 6:10 PM

The

context.get

context.get

context.get

context.get might be a

hono

hono

hono

hono thing, i forgot, either way. I'm not sure how to access my variables that i set from my middlewares

Llukas Yes, but am i able to do `context.get('variable')` within the queue? My CTX is e...

M

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

I don't think that makes sense, since Hono isn't thinking about a queue consumer. What variable is being set? Where would it be set?

MMatt Silverlock I don't think that makes sense, since Hono isn't thinking about a queue consumer...

L

lukasOP•5/21/24, 6:51 PM

Ah thats right, so when a queue consumer consumes, its not apart of the

fetch

fetch

handler?

L

lukasOP•5/21/24, 6:51 PM

Meaning, no middleware is actually ran for the queue consumer. youd have to forward headers to the consumer for any bearer tokens, etc if you need to check that

Llukas Meaning, no middleware is actually ran for the queue consumer. youd have to forw...

H

Hello, I’m Allie!•5/21/24, 7:43 PM

How are you connecting it to Hono? Hono shouldn’t be called at all in a Queue Consumer, Middleware or otherwise

H

Hello, I’m Allie!•5/21/24, 7:44 PM

And IMO there’s no real reason to use Hono for the Consumer portion anyway(unless you are passing Requests back directly into the Queue)

H

Hello, I’m Allie!•5/21/24, 7:44 PM

Which seems very odd

L

lukasOP•5/21/24, 7:46 PM

export const app = new Hono<HonoContext>();

export default {
  async fetch(request: Request, env: TEnv, ctx: ExecutionContext) {
    const validatedEnv = envSchema.safeParse(env);

    if (!validatedEnv.success) {
      console.log(validatedEnv.error);
      return Response.json({ message: "Invalid ENV variables." });
    }

    return app.fetch(request, validatedEnv.data, ctx);
  },
  async queue() {...}
};

export const app = new Hono<HonoContext>();

export default {
  async fetch(request: Request, env: TEnv, ctx: ExecutionContext) {
    const validatedEnv = envSchema.safeParse(env);

    if (!validatedEnv.success) {
      console.log(validatedEnv.error);
      return Response.json({ message: "Invalid ENV variables." });
    }

    return app.fetch(request, validatedEnv.data, ctx);
  },
  async queue() {...}
};

export const app = new Hono<HonoContext>();

export default {
  async fetch(request: Request, env: TEnv, ctx: ExecutionContext) {
    const validatedEnv = envSchema.safeParse(env);

    if (!validatedEnv.success) {
      console.log(validatedEnv.error);
      return Response.json({ message: "Invalid ENV variables." });
    }

    return app.fetch(request, validatedEnv.data, ctx);
  },
  async queue() {...}
};

export const app = new Hono<HonoContext>();

export default {
  async fetch(request: Request, env: TEnv, ctx: ExecutionContext) {
    const validatedEnv = envSchema.safeParse(env);

    if (!validatedEnv.success) {
      console.log(validatedEnv.error);
      return Response.json({ message: "Invalid ENV variables." });
    }

    return app.fetch(request, validatedEnv.data, ctx);
  },
  async queue() {...}
};

I just thought i'd be able to access some in-memory context set by my fetch middlewares, but the queue consumer is isolated from that, so I just created a DB client manually to do operations within the consumer.

My flow is

Web -> route endpoint -> queue producer -> queue consumer -> process messages

Web -> route endpoint -> queue producer -> queue consumer -> process messages

Web -> route endpoint -> queue producer -> queue consumer -> process messages

Web -> route endpoint -> queue producer -> queue consumer -> process messages

Llukas ```js export const app = new Hono<HonoContext>(); export default { async fetc...

H

Hello, I’m Allie!•5/21/24, 7:48 PM

Note too, unless you allow others access to deploy Workers to your account, the Env should always have the shape you expect, and thus shouldn’t need to be validated

HHello, I’m Allie!Note too, unless you allow others access to deploy Workers to your account, the ...

L

lukasOP•5/21/24, 7:49 PM

You make a good point, if configured correctly and deployed only by me as the owner, no point in validating it

S

Sameer Ali•5/21/24, 8:22 PM

in producer worker I am getting

workers.api.error.queue_handler_missing

workers.api.error.queue_handler_missing

workers.api.error.queue_handler_missing

workers.api.error.queue_handler_missing error when deploying. the producer worker doesn't handle the queue and in wrnagler.toml i have only producer part of the configuration.

F

Fokke•5/21/24, 10:37 PM

I am using cloudflare queues to build a data pipeline. My goal was to push a bunch of messages in queues and have workes process them as quickly as possible. However hitting some limitations and wondering if there is a way around this.

I want processing to happen as fast as possible and so ideally my workers scale up when there is more than 1 message in the queue. I don't mind paying a little extra for additional workers.

I know there is a way to configure the maximum number of consumers, but is there a way where we can define a minimum number of consuming workers?

FFokke I am using cloudflare queues to build a data pipeline. My goal was to push a bun...

I

Ian•5/21/24, 11:53 PM

specifying a minimum doesn't really make sense. It should just start assigning more consumers if there are enough messages. Make sure your

max_batch_size

max_batch_size

is set to

so that it doesn't assign a batch of, say, 5 messages to a single worker.

W

Whisker•5/22/24, 7:26 AM

hi guys, does anyone know how to debug a shared queue function in multiple worker projects locally. I want to use the same queue, producer in project A, consumer is project B, thanks.

WWhisker hi guys, does anyone know how to debug a shared queue function in multiple worke...

I

Ian•5/22/24, 9:14 AM

What have you tried and what issues are you running into?

I

itecheck•5/23/24, 2:00 AM

你有什么

C

cf-workers•5/23/24, 3:51 AM

Hi there, I am using workers and queues to process large no of events. I have few questions
1. Is there a way to monitor queue performance and raise an alert in case of any issues? Same for workers?
2. Is there a way I can setup logs for queues and workers? I see an option to upload the logs to r2, would like to know if there is better option to access then and perform search quickly.
Thank you for your help

I

Ian•5/23/24, 6:06 AM

I have a monorepo (managed by LernaJS) that has a cloudflare pages package and a queue consumer package. When I start both packages in dev mode they both seem to set up their own

.wrangler

.wrangler

.wrangler

.wrangler directories and seem to be isolated, so they don't share a queue. To make them able to talk to the same queue do I need to use

--persist-to

--persist-to

--persist-to

--persist-to on both?

I

Ian•5/23/24, 6:48 AM

or should I be looking into using miniflare directly?

Ccf-workers Hi there, I am using workers and queues to process large no of events. I have fe...

S

Sam•5/23/24, 7:22 AM

Use their newly acquired product #baselime for logging.

J

JinBlack•5/23/24, 1:56 PM

"Will Cloudflare consider adding support for sharded partitioning in their queue service in the future?

Eengrtitus Is retry({delaySeconds: ....}) working for anyone ? The message gets retried im...

M

Matt Silverlock•5/24/24, 12:36 PM

Can you share your code and exactly how you are calling the API?

JJinBlack "Will Cloudflare consider adding support for sharded partitioning in their queue...

M

Matt Silverlock•5/24/24, 12:37 PM

Do you mean “batching based on a user-defined partitioning key?”

Eengrtitus Hi, it's called like so ``` const { body } = message; if (body) { const { req...

M

Matt Silverlock•5/25/24, 6:45 PM

Is your Worker throwing errors? That would cause a retry before you call

delaySeconds

delaySeconds

delaySeconds

delaySeconds

Eengrtitus Hi, it's called like so ``` const { body } = message; if (body) { const { req...

M

Matt Silverlock•5/25/24, 6:45 PM

Your queues dashboard metrics will show if you have errors

X

XATEV•5/26/24, 7:55 AM

Has anyone here tried to integrate CF queues with logstash? What would be the best approach for that, just use the http_poll input?

O

Ohad•5/26/24, 8:16 AM

is there any

rust

rust

rust

rust based

queue

queue

i can use as reference?

O

Ohad•5/26/24, 8:19 AM

nvm found it https://github.com/cloudflare/workers-rs/blob/main/examples/queue/src/lib.rs

GitHub

workers-rs/examples/queue/src/lib.rs at main · cloudflare/workers-rs

Write Cloudflare Workers in 100% Rust via WebAssembly - cloudflare/workers-rs

Eengrtitus No, it isn't. We catch and log all errors. Our workaround now is to ack the old...

M

Matt Silverlock•5/26/24, 11:36 AM

Can you share your account ID & queue ID? This sounds like a bug if this isn’t working (and would be the first report, hence my questions!)

Eengrtitus Oh....got it Account Id: 6caf95d4a4a1a9be07f3aff69662c84c QueueId: 02b291e03c06...

M

Matt Silverlock•5/26/24, 2:00 PM

Thanks. I’d still recommend wrapping your fetch call with a try-catch so the delay is set even if an exception is thrown.

O

Ohad•5/26/24, 5:42 PM

What’s the recommended way to use queues for logs in rust ? As in, dynamic content .
Currently I’m converting back and forth between serde_json::Value and JsValue. But it feels very dirty.

S

shreeyansb•5/27/24, 7:59 AM

Hi I am trying to test the delay functionality of Queues and I am unable to get it to work. The msg is consumed almost instantly.

export const producerTest = async (c) => {
  try {
      await c.env.TestQueue.send(JSON.stringify({
      slug: 'test123',
      timestamp: new Date().toISOString()
    }), { delaySeconds: 300 });
    return c.json('Queued', httpStatus.OK);
  } catch (error) {
    console.log(`Error in test ${error.message}`);
    const response = createResponse(
      {},
      {
        message: `Error occured in test ${error.message}`,
      }
    );
    return c.json(response, httpStatus.INTERNAL_SERVER_ERROR);
  }
};

export const producerTest = async (c) => {
  try {
      await c.env.TestQueue.send(JSON.stringify({
      slug: 'test123',
      timestamp: new Date().toISOString()
    }), { delaySeconds: 300 });
    return c.json('Queued', httpStatus.OK);
  } catch (error) {
    console.log(`Error in test ${error.message}`);
    const response = createResponse(
      {},
      {
        message: `Error occured in test ${error.message}`,
      }
    );
    return c.json(response, httpStatus.INTERNAL_SERVER_ERROR);
  }
};

export const producerTest = async (c) => {
  try {
      await c.env.TestQueue.send(JSON.stringify({
      slug: 'test123',
      timestamp: new Date().toISOString()
    }), { delaySeconds: 300 });
    return c.json('Queued', httpStatus.OK);
  } catch (error) {
    console.log(`Error in test ${error.message}`);
    const response = createResponse(
      {},
      {
        message: `Error occured in test ${error.message}`,
      }
    );
    return c.json(response, httpStatus.INTERNAL_SERVER_ERROR);
  }
};

export const producerTest = async (c) => {
  try {
      await c.env.TestQueue.send(JSON.stringify({
      slug: 'test123',
      timestamp: new Date().toISOString()
    }), { delaySeconds: 300 });
    return c.json('Queued', httpStatus.OK);
  } catch (error) {
    console.log(`Error in test ${error.message}`);
    const response = createResponse(
      {},
      {
        message: `Error occured in test ${error.message}`,
      }
    );
    return c.json(response, httpStatus.INTERNAL_SERVER_ERROR);
  }
};

Sshreeyansb Hi I am trying to test the delay functionality of Queues and I am unable to get ...

M

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

Can you share your account ID & queue ID?

MMatt Silverlock Can you share your account ID & queue ID?

S

shreeyansb•5/27/24, 3:23 PM

Account id: 5434f32df6958dd7dbb3c6a39412f605
Queue id: 23f868690fee4853a4e62a2243c293e6

Sshreeyansb Account id: 5434f32df6958dd7dbb3c6a39412f605 Queue id: 23f868690fee4853a4e62a224...

M

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

Thanks! This is the second report: we’re looking into it

M

Matt Silverlock•5/27/24, 5:25 PM

Note: so far I cannot reproduce this, with a regular Worker consumer calling

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90}) - messages are only delivered 90s later. The

batch_timeout

batch_timeout

batch_timeout

batch_timeout on this queue is 30s.

M

Matt Silverlock•5/27/24, 5:28 PM

Consumer handles the messages, calls

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90})

msg.retry({delaySeconds: 90}) -> messages are only re-delivered ~90s later. No sooner. Not immediately.

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:25:37 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 121500 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 120635 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 117075 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 120043 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 119587 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 119158 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 118721 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 118457 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 118139 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 117884 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 117632 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:25:37 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 121500 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 120635 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 117075 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 120043 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 119587 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 119158 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 118721 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 118457 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 118139 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 117884 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 117632 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:25:37 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 121500 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 120635 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 117075 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 120043 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 119587 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 119158 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 118721 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 118457 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 118139 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 117884 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 117632 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:25:37 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 121500 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 120635 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 117075 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 120043 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 119587 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 119158 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 118721 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 118457 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 118139 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 117884 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 117632 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:27:08 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 212948 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 212083 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 209080 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 208523 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 211491 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 211035 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 210606 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 210169 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 209905 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 209587 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 209332 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:27:08 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 212948 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 212083 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 209080 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 208523 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 211491 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 211035 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 210606 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 210169 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 209905 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 209587 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 209332 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:27:08 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 212948 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 212083 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 209080 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 208523 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 211491 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 211035 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 210606 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 210169 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 209905 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 209587 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 209332 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}

Queue test-queue (11 messages) - Ok @ 5/27/2024, 1:27:08 PM
  (log) reading 11 messages from test-queue
  (log) 2e8b3c1741fe1a8dadab995ba4576bea (latency: 212948 ms | retried: true): {"path":"/","timestamp":1716830615990,"eyeballColo":"EWR","shouldDelay":true}
  (log) 2eb086a6c0543b5eeee7c93c096c6614 (latency: 212083 ms | retried: true): {"path":"/","timestamp":1716830616869,"eyeballColo":"EWR","shouldDelay":true}
  (log) 3f93615324c32404c594f4aeabf22f1c (latency: 209080 ms | retried: true): {"path":"/","timestamp":1716830619872,"eyeballColo":"EWR","shouldDelay":true}
  (log) ebb3df2f9730b68e271de85bb0aec390 (latency: 208523 ms | retried: true): {"path":"/","timestamp":1716830620424,"eyeballColo":"EWR","shouldDelay":true}
  (log) 073d71fc8c745efd43ae95d7d575a2a4 (latency: 211491 ms | retried: true): {"path":"/","timestamp":1716830617459,"eyeballColo":"EWR","shouldDelay":true}
  (log) d7383b6412b6373c10b3dae024031df0 (latency: 211035 ms | retried: true): {"path":"/","timestamp":1716830617910,"eyeballColo":"EWR","shouldDelay":true}
  (log) dc8b7593178b7e1f7abbc40894d3cddc (latency: 210606 ms | retried: true): {"path":"/","timestamp":1716830618344,"eyeballColo":"EWR","shouldDelay":true}
  (log) 1b85ca96e8c0bbd7778b8e2640dd9d86 (latency: 210169 ms | retried: true): {"path":"/","timestamp":1716830618779,"eyeballColo":"EWR","shouldDelay":true}
  (log) 8ffb38f375ecd59610aaeb14fb28c587 (latency: 209905 ms | retried: true): {"path":"/","timestamp":1716830619038,"eyeballColo":"EWR","shouldDelay":true}
  (log) 138d3c0b419d12b7611dd0fcf89e5d40 (latency: 209587 ms | retried: true): {"path":"/","timestamp":1716830619363,"eyeballColo":"EWR","shouldDelay":true}
  (log) 7df01f266ac7c679953db172cb88e42d (latency: 209332 ms | retried: true): {"path":"/","timestamp":1716830619617,"eyeballColo":"EWR","shouldDelay":true}

M

Matt Silverlock•5/27/24, 5:30 PM

For those playing at home, my consumer is very simple: messages that have a

shouldDelay: true

shouldDelay: true

shouldDelay: true

shouldDelay: true property get retried with a fixed delay.

max_retries

max_retries

is 10 so they eventually get deleted:

async queue(batch: MessageBatch<Message>, env: Env, ctx: ExecutionContext) {
    console.log(`reading ${batch.messages.length} messages from ${batch.queue}`);
    for (const msg of batch.messages) {
        let latency = Date.now() - msg.timestamp.valueOf();

        let retried = false;
        if (msg.body.shouldDelay) {
            msg.retry({ delaySeconds: 90 });
            retried = true;
        } else {
            msg.ack();
        }

        console.log(`${msg.id} (latency: ${latency} ms | retried: ${retried}): ${JSON.stringify(msg.body)}`);
    }
}

async queue(batch: MessageBatch<Message>, env: Env, ctx: ExecutionContext) {
    console.log(`reading ${batch.messages.length} messages from ${batch.queue}`);
    for (const msg of batch.messages) {
        let latency = Date.now() - msg.timestamp.valueOf();

        let retried = false;
        if (msg.body.shouldDelay) {
            msg.retry({ delaySeconds: 90 });
            retried = true;
        } else {
            msg.ack();
        }

        console.log(`${msg.id} (latency: ${latency} ms | retried: ${retried}): ${JSON.stringify(msg.body)}`);
    }
}

async queue(batch: MessageBatch<Message>, env: Env, ctx: ExecutionContext) {
    console.log(`reading ${batch.messages.length} messages from ${batch.queue}`);
    for (const msg of batch.messages) {
        let latency = Date.now() - msg.timestamp.valueOf();

        let retried = false;
        if (msg.body.shouldDelay) {
            msg.retry({ delaySeconds: 90 });
            retried = true;
        } else {
            msg.ack();
        }

        console.log(`${msg.id} (latency: ${latency} ms | retried: ${retried}): ${JSON.stringify(msg.body)}`);
    }
}

async queue(batch: MessageBatch<Message>, env: Env, ctx: ExecutionContext) {
    console.log(`reading ${batch.messages.length} messages from ${batch.queue}`);
    for (const msg of batch.messages) {
        let latency = Date.now() - msg.timestamp.valueOf();

        let retried = false;
        if (msg.body.shouldDelay) {
            msg.retry({ delaySeconds: 90 });
            retried = true;
        } else {
            msg.ack();
        }

        console.log(`${msg.id} (latency: ${latency} ms | retried: ${retried}): ${JSON.stringify(msg.body)}`);
    }
}

Eengrtitus Oh..Not the experience here. It gets retried immediately. Not sure if this matt...

M

Matt Silverlock•5/27/24, 8:11 PM

Did it work previously for you at all?

send

send

send

send vs

sendBatch

sendBatch

sendBatch

sendBatch should make no difference.

Am i able to access my env context variables (they get set in middleware) in a queue consumer? I wan

Similar Threads

Similar Threads

Similar Threads