@Matt any news/ updates? - Cloudflare Developers

© 2026 Hedgehog Software, LLC

More

Communities Docs About Terms Privacy

Ccg @Matt any news/ updates?

MattIPv4

M

MattIPv4•11/24/23, 5:18 AM

I feel you pinged the wrong Matt

C

cgOP•11/24/23, 5:18 AM

ah shoot, sorry mate

C

cgOP•11/24/23, 5:19 AM

well, happy thanksgiving

C

cgOP•11/24/23, 5:19 AM

nice work

C

cgOP•11/24/23, 5:19 AM

love what you're doing

dattaproffs

Ddattaproffs Hi, i don't see anything about it in the docs but i thought i could ask anyway. ...

Mark

M

Mark•11/24/23, 10:31 AM

I have also been looking for the ability to delay queue messages. Is this possible?

Mark

MMark I have also been looking for the ability to delay queue messages. Is this possib...

Matt Silverlock

M

Matt Silverlock•11/24/23, 2:26 PM

dattaproffs

Ddattaproffs Hi, i don't see anything about it in the docs but i thought i could ask anyway. ...

Matt Silverlock

M

Matt Silverlock•11/24/23, 2:27 PM

See the pinned messages: queues

Matt Silverlock

MMatt Silverlock See the pinned messages: https://discord.com/channels/595317990191398933/1008691...

dattaproffs

D

dattaproffs•11/24/23, 6:12 PM

Ah ofc. Thanks

amg

A

amg•11/26/23, 4:49 PM

Hi, I am not able to see the queue page to create a queue. The website always redirect me to the overview route

amg

A

amg•11/26/23, 4:52 PM

I am able to create the queue with wrangler but I am always redirect to this page

amg

A

amg•11/26/23, 4:53 PM

amg

A

amg•11/26/23, 4:59 PM

It seems that until you create a worker, you can not access to the queues

C

cgOP•11/26/23, 10:54 PM

@Matt
Any news / updates on the Service Temporarily Unavailable issue?

Ccg @Matt Any news / updates on the Service Temporarily Unavailable issue?

Matt Silverlock

M

Matt Silverlock•11/26/23, 11:08 PM

Not yet. Are you still seeing this repeatedly per your repro above?

C

cgOP•11/26/23, 11:08 PM

yep, still seing this

C

cgOP•11/26/23, 11:26 PM

I can consistently reproduce it with at least one queue message type

C

cgOP•11/26/23, 11:26 PM

The code is rather simple IMO, so I'm not really sure what to do

Ccg The code is rather simple IMO, so I'm not really sure what to do

Matt Silverlock

M

Matt Silverlock•11/26/23, 11:52 PM

We won’t know until we investigate further. We’ve not seen a case like this before where the queue is returning hard errors every N messages.

C

cgOP•11/26/23, 11:53 PM

I'm available to assist with any investigate if it would help, I'm in Cali time for the next week, then back to AEDT

Shihmin Lee

S

Shihmin Lee•11/27/23, 4:00 AM

Hey guys. A question about dead_letter_queue. I noticed the message gets encrypted when it goes to dead_letter_queue. Is there a way to send the message as JSON instead?

Shihmin Lee

S

Shihmin Lee•11/27/23, 1:43 PM

Another question is for message inside the dead_letter_queue. Is there a way to move it into the original queue manually on the GUI?

Shihmin Lee

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

Matt Silverlock

M

Matt Silverlock•11/27/23, 2:38 PM

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.

Shihmin Lee

S

Shihmin Lee•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?

kian

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

Shihmin Lee

S

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

Thanks.

Shihmin Lee

S

Shihmin Lee•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.

Unsmart

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

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

Shihmin Lee

S

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

Gotcha.

C

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

Matt Silverlock

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

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

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

C

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

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

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

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

C

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

C

cgOP•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' })),
      );
    }),
  );
};

kian

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.

Unsmart

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

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

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

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

Which feels happily within the 1000 request limit

kian

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

C

cgOP•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_concurrentmax_concurrent is default

auto

auto

Ccg Which feels happily within the 1000 request limit

kian

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

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

kian

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

C

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

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

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