Hi there, I am using workers and queues to process large no of events. I have few questions 1. Is th

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

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

vs

sendBatch

sendBatch

sendBatch

sendBatch should make no difference.

MMatt Silverlock Did it work previously for you at all? `send` vs `sendBatch` should make no diff...

S

shreeyansb•5/28/24, 4:00 AM

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

M

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!

S

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

M

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

Per queues - are you also using this locally?

F

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

M

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.

G

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

M

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

Do you mean pull consumer?

M

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?

G

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

M

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.

S

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

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.

P

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...?

P

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}` });
});

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)}`
            );
        }
    },
};

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)}`
            );
        }
    },
};

P

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

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

L

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? ...

M

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

L

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.

Hi there, I am using workers and queues to process large no of events. I have few questions 1. Is th

Similar Threads

Similar Threads

Similar Threads