Another question is for message inside the dead_letter_queue. Is there a way to move it into the ori

SShihmin Lee Another question is for message inside the dead_letter_queue. Is there a way to ...

M

You cannot move a message "back" to the original queue. You can set up the original queue as the DLQ for the "DLQ" you have, but you cannot move individual messages.

S

Shihmin LeeOP•11/27/23, 9:56 PM

I see. Is there a way to use the existing worker to subscribe to 2 queues at the same time? It might seem strange if I need another worker for this DLQ and maybe I also need a DLQ DLQ worker for this DLQ? :p

As for the JSON part. Is there a way to see the message as JSON in the DLQ?

K

kian•11/27/23, 10:11 PM

Is there a way to use the existing worker to subscribe to 2 queues at the same time?

Yep https://developers.cloudflare.com/queues/learning/how-queues-works/#create-a-consumer

S

Shihmin LeeOP•11/27/23, 10:17 PM

Thanks.

S

Shihmin LeeOP•11/27/23, 10:44 PM

Hey Kian, maybe I wasn't clear enough in my previous request. Can you provide a document that shows how to configure a single worker to subscribe to two different queues with two consumers? I can't seem to figure out how to configure [[queues.consumers]] in the .toml file.

U

Unsmart•11/27/23, 10:48 PM

The docs he linked is how you do that.

[[queues.consumers]]
queue = "1"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "2"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "1"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "2"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "1"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "2"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "1"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

[[queues.consumers]]
queue = "2"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional

export default {
  async queue(batch: MessageBatch<Error>, env: Environment): Promise<void> {
    switch (batch.queue) {
      case '1':
        // handle queue 1
        break;
      case '2':
        // handle queue 2
        break;
    }
  },
};

export default {
  async queue(batch: MessageBatch<Error>, env: Environment): Promise<void> {
    switch (batch.queue) {
      case '1':
        // handle queue 1
        break;
      case '2':
        // handle queue 2
        break;
    }
  },
};

export default {
  async queue(batch: MessageBatch<Error>, env: Environment): Promise<void> {
    switch (batch.queue) {
      case '1':
        // handle queue 1
        break;
      case '2':
        // handle queue 2
        break;
    }
  },
};

export default {
  async queue(batch: MessageBatch<Error>, env: Environment): Promise<void> {
    switch (batch.queue) {
      case '1':
        // handle queue 1
        break;
      case '2':
        // handle queue 2
        break;
    }
  },
};

S

Shihmin LeeOP•11/27/23, 10:54 PM

Gotcha.

C

cg•11/28/23, 12:29 AM

@Matt, apologies on keeping @'ing you but I've heard nothing from the support ticket.
If there any time lines or information you can provide, that would be really helpful. I'll have to start looking at other services to accomodate the Queue, like moving to SQS as it's starting to impact business

Ccg @Matt, apologies on keeping @'ing you but I've heard nothing from the support ti...

M

Matt Silverlock•11/28/23, 12:32 AM

Unfortunately we’re still investigating. I hate to have this take so long but Queues is still in beta, and so issues like this (which have been rare) may still crop up. I also assume you’ve created a fresh queue and still reproduce this.

C

cg•11/28/23, 12:55 AM

Thanks, I appreciate the reply. Yea, a fresh queue still had the issue.

C

cg•11/28/23, 1:26 AM

Could I confirm that when calling

QUEUE.send()

QUEUE.send()

within a consumer, that it doesn't spawn a new worker within the same "thread"

C

cg•11/28/23, 1:26 AM

I'm seeing

Queue sendBatch failed: Subrequest depth limit exceeded.  This request recursed through Workers too many times.

Queue sendBatch failed: Subrequest depth limit exceeded.  This request recursed through Workers too many times.

but the message that has invoked the consumer should not reach this limit at all

C

cg•11/28/23, 1:28 AM

This is the code that is being hit (has been modified)

C

cg•11/28/23, 1:29 AM

export const handleMessage = async ({
  message,
  env,
}) => {
  const { items, nextPageToken } =
    await api.listByStatus({
      limit: 100,
      nextPageToken: message.nextPageToken,
      status: 'approved',
    });

  if (nextPageToken) {
    await sendQueueMessage({
      message: {
        type: '<REDACTED>',
        nextPageToken,
      },
      env,
    });
  }

  const toBeProcessed = items
    .filter((f) => !f.processed)

  await sendBatchQueueMessage({
    batch: toBeProcessed.map((transaction) => ({
      message: {
        transaction,
        type: '<OTHERMESSAGE>',
      },
    })),
    env,
  });
};

export const handleMessage = async ({
  message,
  env,
}) => {
  const { items, nextPageToken } =
    await api.listByStatus({
      limit: 100,
      nextPageToken: message.nextPageToken,
      status: 'approved',
    });

  if (nextPageToken) {
    await sendQueueMessage({
      message: {
        type: '<REDACTED>',
        nextPageToken,
      },
      env,
    });
  }

  const toBeProcessed = items
    .filter((f) => !f.processed)

  await sendBatchQueueMessage({
    batch: toBeProcessed.map((transaction) => ({
      message: {
        transaction,
        type: '<OTHERMESSAGE>',
      },
    })),
    env,
  });
};

export const handleMessage = async ({
  message,
  env,
}) => {
  const { items, nextPageToken } =
    await api.listByStatus({
      limit: 100,
      nextPageToken: message.nextPageToken,
      status: 'approved',
    });

  if (nextPageToken) {
    await sendQueueMessage({
      message: {
        type: '<REDACTED>',
        nextPageToken,
      },
      env,
    });
  }

  const toBeProcessed = items
    .filter((f) => !f.processed)

  await sendBatchQueueMessage({
    batch: toBeProcessed.map((transaction) => ({
      message: {
        transaction,
        type: '<OTHERMESSAGE>',
      },
    })),
    env,
  });
};

export const handleMessage = async ({
  message,
  env,
}) => {
  const { items, nextPageToken } =
    await api.listByStatus({
      limit: 100,
      nextPageToken: message.nextPageToken,
      status: 'approved',
    });

  if (nextPageToken) {
    await sendQueueMessage({
      message: {
        type: '<REDACTED>',
        nextPageToken,
      },
      env,
    });
  }

  const toBeProcessed = items
    .filter((f) => !f.processed)

  await sendBatchQueueMessage({
    batch: toBeProcessed.map((transaction) => ({
      message: {
        transaction,
        type: '<OTHERMESSAGE>',
      },
    })),
    env,
  });
};

C

cg•11/28/23, 1:31 AM

These are the two helper functions:

export const sendQueueMessage = ({ message, env }: SendQueueMessage) => {
  return env.PROCESS_QUEUE.send(message, { contentType: 'json' });
};

export const sendBatchQueueMessage = ({
  batch,
  env,
}: SendBatchQueueMessage) => {
  const ch = chunk(batch, 20);
  return Promise.all(
    ch.map((c) => {
      return env.PROCESS_QUEUE.sendBatch(
        c.map((body) => ({ body: body.message, contentType: 'json' })),
      );
    }),
  );
};

export const sendQueueMessage = ({ message, env }: SendQueueMessage) => {
  return env.PROCESS_QUEUE.send(message, { contentType: 'json' });
};

export const sendBatchQueueMessage = ({
  batch,
  env,
}: SendBatchQueueMessage) => {
  const ch = chunk(batch, 20);
  return Promise.all(
    ch.map((c) => {
      return env.PROCESS_QUEUE.sendBatch(
        c.map((body) => ({ body: body.message, contentType: 'json' })),
      );
    }),
  );
};

export const sendQueueMessage = ({ message, env }: SendQueueMessage) => {
  return env.PROCESS_QUEUE.send(message, { contentType: 'json' });
};

export const sendBatchQueueMessage = ({
  batch,
  env,
}: SendBatchQueueMessage) => {
  const ch = chunk(batch, 20);
  return Promise.all(
    ch.map((c) => {
      return env.PROCESS_QUEUE.sendBatch(
        c.map((body) => ({ body: body.message, contentType: 'json' })),
      );
    }),
  );
};

export const sendQueueMessage = ({ message, env }: SendQueueMessage) => {
  return env.PROCESS_QUEUE.send(message, { contentType: 'json' });
};

export const sendBatchQueueMessage = ({
  batch,
  env,
}: SendBatchQueueMessage) => {
  const ch = chunk(batch, 20);
  return Promise.all(
    ch.map((c) => {
      return env.PROCESS_QUEUE.sendBatch(
        c.map((body) => ({ body: body.message, contentType: 'json' })),
      );
    }),
  );
};

K

kian•11/28/23, 2:02 AM

Concurrent consumers are in the same Worker 'instance' I believe, so you can sometimes run into limits that you wouldn't expect to hit.

U

Unsmart•11/28/23, 4:26 AM

I'm guessing durable object alarms dont have this subrequest depth tracking? I think quite a few people use it for crons so could use that for the infinite loop things

C

cg•11/28/23, 6:07 AM

Oh interesting - I'm not really sure how to deal with this though. I could try setting this:

await api.listByStatus({
      limit: 100,

await api.listByStatus({
      limit: 100,

to something like limit 10 or limit 5

C

cg•11/28/23, 6:08 AM

But from looking at the code in it's current state, I'd only expect like 6 requests (4 batches of 20 + maybe two others)

C

cg•11/28/23, 6:08 AM

Which feels happily within the 1000 request limit

Kkian Concurrent consumers are in the same Worker 'instance' I believe, so you can som...

C

cg•11/28/23, 6:12 AM

Oh, well, I must say that limitation kinda sucks - trying to juggle the number of requests a worker can make is tricky enough without also having to consider the possibility of this worker being run concurrently and accounting for those workers. Especially considering that the

max_concurrent

max_concurrent

is default

auto

auto

Ccg Which feels happily within the 1000 request limit

K

kian•11/28/23, 6:12 AM

Depth is a different limit than exceeding the 50/1000 subrequests limits - it's not really documented and is generally only caused by recursion.

C

cg•11/28/23, 6:13 AM

I've had to set my max_batch to be

because I've already come into trouble with too many messages being handled by 1 workers and hitting subrequest limits

Kkian Depth is a different limit than exceeding the 50/1000 subrequests limits - it's ...

C

cg•11/28/23, 6:15 AM

Right, well, I guess I'm a bit lost as to what to do. We will have to see what @AlexRobinson comes up with tomorrow

C

cg•11/28/23, 6:19 AM

Am I using queues incorrectly?
Trying to paginate through data from an API felt like Queues were a good fit

U

Unsmart•11/28/23, 6:46 AM

Durable objects + alarms would be better for that imo

C

cg•11/28/23, 6:49 AM

mmm, I don't really need to global-ness that DO offers

C

cg•11/28/23, 6:50 AM

but I'll explore it, ty

Ccg Am I using queues incorrectly? Trying to paginate through data from an API felt...

M

Matt Silverlock•11/28/23, 1:47 PM

This is a sharp edge we should just fix.

S

Shihmin LeeOP•11/28/23, 9:35 PM

Hey guys. I've also created an issue about the queue not logging, processing jobs, or stopping unexpectedly without generating logs. Would anyone have insight into what might be causing this behavior? https://github.com/cloudflare/workers-sdk/issues/4517

GitHub

🐛 BUG: Queue are not working correctly when you have 2 consumers · ...

Which Cloudflare product(s) does this pertain to? Queues What version(s) of the tool(s) are you using? 3.17.1 What version of Node are you using? 20.10.0 What operating system are you using? Mac De...

S

Shihmin LeeOP•11/29/23, 3:25 AM

I have another question. When I am consuming the queue messages, I notice that sometimes I find myself in a scenario where both the queue and kv are failing. These are the logs I receive through Logpush / Datadog. Do you have any suggestions on how this could happen and how do I get around it?

K

kian•11/29/23, 9:16 AM

Sounds like the same issue as the conversation above.

SShihmin Lee I have another question. When I am consuming the queue messages, I notice that s...

M

Matt Silverlock•11/29/23, 12:28 PM

How long have you been using Queues? Trying to understand if you just set it up and saw this, or if you saw this after using Queues for some time.

D

drk•11/29/23, 1:18 PM

are there good examples of a pages/workers/queue monorepo? i deploy a remix project to pages currently per cloudflare remix guides and want to leverage queues. so far it seems like a single queue is the only documented setup and given remix+pages constraints that makes sense, but i'm curious about additional workers as consumers and my remix app as a producer primarily.

D

drk•11/29/23, 1:19 PM

and specifically running multiple workers/consumers locally so i can develop end-to-end, but i think i read that you can only run a single queue consumer at a time locally?

D

drk•11/29/23, 1:20 PM

also looking for tips on general dev workflow for a setup like this. like if it's best to work on consumers in isolation, deploy those (or preview deploy them for dev purposes) and have my remix app running locally use a dev specific queue that's live in cloudflare.

MMatt Silverlock How long have you been using Queues? Trying to understand if you just set it up ...

S

Shihmin LeeOP•11/29/23, 1:53 PM

I am just using Queues for a few days. I suspect this is me hitting some rate limits? I also notice a similar issue for Queue + D1.

S

Shihmin LeeOP•11/29/23, 1:59 PM

Although I doubt that I was hitting the rate limits since I am not seeing 429, also I only have 2 workers on 2 queue with 1 consumer each inserted like 1000 records over 5 mins on avg.

SShihmin Lee I am just using Queues for a few days. I suspect this is me hitting some rate li...

M

Matt Silverlock•11/29/23, 2:01 PM

Your issue appears identical to queues

S

Shihmin LeeOP•11/29/23, 2:01 PM

Gotcha. I've never seen Subrequest depth limit error message though.

SShihmin Lee Gotcha. I've never seen Subrequest depth limit error message though.

M

Matt Silverlock•11/29/23, 4:23 PM

Per that chat message, that error does not always manifest.

TTyler Benfield Is there any known degradation to queues in the last ~12 hours? We are seeing so...

M

Matt Silverlock•11/30/23, 1:14 PM

Not specific to the last 12 hours, but we have had a couple of reports about consumers not being invoked. We’re looking into the root cause.

TTyler Benfield Thanks @Matt! Let us know if there's any information we can provide. We're going...

M

Matt Silverlock•11/30/23, 1:36 PM

Note that it’s not related to concurrency - that just makes it more obvious.

If you have an account ID and queue ID we can add to our list.

MMatt Silverlock Per that chat message, that error does not always manifest.

S

Shihmin LeeOP•11/30/23, 3:33 PM

Hey Matt. So do you know if there's a timeline that this is going to be addressed? I'm interested in moving my infra from S3 to using R2, D1, Queues and KVs and such. But so far it's been harsh to see that I am getting so many 503s for combinations of kv, queue, d1, ... etc in the queue msg as well as all the immediate messages that got consumed.

Another question is for message inside the dead_letter_queue. Is there a way to move it into the ori

Similar Threads