it's `89a515eb79...`

Ccg it's `89a515eb79...`

M

FWIW I don't see a ton of messages sent here — at least in the underlying Queue broker. is that

Service Temporarily Unavailable

Service Temporarily Unavailable

coming as an exception directly from

sendBatch

sendBatch

? (a code snippet would be helpful)

C

cgOP•11/23/23, 7:49 PM

Yea, I really turned down the number of events being sent since the backlog of 200k yesterday

MMatt Silverlock FWIW I don't see a ton of messages sent here — at least in the underlying Queue ...

C

cgOP•11/23/23, 7:50 PM

This is the function that I'm invoking:

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

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

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

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

C

cgOP•11/23/23, 7:50 PM

chunk

chunk

here is

lodash

lodash

's chunk function, to break messages into groups

Ccg This is the function that I'm invoking: ``` export const sendBatchQueueMessage ...

M

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

I suspect you're potentially hitting connection limits before you hit the Queues service here because of

Promise.all

Promise.all

especially if you're running this inside your consumer. Something like
```
p-map
```
```
p-map
```
(https://github.com/sindresorhus/p-map) might help to keep the concurrency level down - you can only have 6 simultaneous connections open.

C

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

Actually, interestingly, this error seems to be coming from this function:

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

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

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

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

C

cgOP•11/23/23, 7:54 PM

oh excuse me, I might be wrong

C

cgOP•11/23/23, 7:54 PM

Thanks, I'll take a look at that

C

cgOP•11/23/23, 7:56 PM

If the 6 connection limit is reached, what happens to the worker?
Is there some error reported somewhere?

Ccg If the 6 connection limit is reached, what happens to the worker? Is there some ...

M

Matt Silverlock•11/23/23, 7:57 PM

Connections will be pending (until they potentially time out). On top of that, you can only make 1000 requests per invocation (e.g. each

send

send

or

sendBatch

sendBatch

counts as one) . An invocation is each call to

queue

queue

or

fetch

fetch

.

C

cgOP•11/23/23, 7:59 PM

I'd be surprised if I am hitting the limit, as the log message that I am observing appears in Google Cloud Log Explorer as a result of a

fetch

fetch

that is finalised at the end of the

queue()

queue()

handler

C

cgOP•11/23/23, 7:59 PM

return handleMessages(batch, { ...env, logger }).finally(() =>
      tracer.finalizeQueueEvent({ batch }),
    );

return handleMessages(batch, { ...env, logger }).finally(() =>
      tracer.finalizeQueueEvent({ batch }),
    );

C

cgOP•11/23/23, 8:01 PM

maybe I'm understanding this wrong, but if I am seeing the log message in Logs Explorer as a result of an outbound

fetch

fetch

, then all the requests within the

queue

queue

invocation must be < 1000, or else the log message would never make it

C

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

I've also seen the error

Queue send failed: Service Temporarily Unavailable

Queue send failed: Service Temporarily Unavailable

C

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

Which I don't believe should have the same batching / concurrency issues

Ccg maybe I'm understanding this wrong, but if I am seeing the log message in Logs E...

M

Matt Silverlock•11/23/23, 8:09 PM

Correct, so we can rule that out.

C

cgOP•11/23/23, 8:44 PM

Well, I must say I'm at a loss here

C

cgOP•11/23/23, 8:44 PM

it might be a bit of a nuclear option, but I could create a new queue?

C

cgOP•11/23/23, 9:01 PM

Well, no, changing to a new Queue did not solve it, still seeing:

 "Queue sendBatch failed: Service Temporarily Unavailable

 "Queue sendBatch failed: Service Temporarily Unavailable

Ccg Well, no, changing to a new Queue did not solve it, still seeing: ``` "Queue s...

M

Matt Silverlock•11/23/23, 9:21 PM

If you set up the most minimal queue - e.g. identical to the tutorial - do you still see the issue? I want to rule out anything else here.

MMatt Silverlock If you set up the most minimal queue - e.g. identical to the tutorial - do you s...

M

Matt Silverlock•11/23/23, 9:21 PM

Heck, just

send(“foo”)

send(“foo”)

C

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

I can send messages via the dashbaord and they get received and processed

C

cgOP•11/23/23, 9:24 PM

The Q seems to be working, it's seems like the consumer fails to produce an event

C

cgOP•11/23/23, 9:28 PM

I'll put this at the top of the handleMessage()

C

cgOP•11/23/23, 9:28 PM

  env.logger.info('Sending foo');
  env.PROCESS_QUEUE.send('FOO')
    .then(() => env.logger.info('sent foo'))
    .catch((e) => env.logger.error('failed to send foo', e));

  env.logger.info('Sending foo');
  env.PROCESS_QUEUE.send('FOO')
    .then(() => env.logger.info('sent foo'))
    .catch((e) => env.logger.error('failed to send foo', e));

  env.logger.info('Sending foo');
  env.PROCESS_QUEUE.send('FOO')
    .then(() => env.logger.info('sent foo'))
    .catch((e) => env.logger.error('failed to send foo', e));

  env.logger.info('Sending foo');
  env.PROCESS_QUEUE.send('FOO')
    .then(() => env.logger.info('sent foo'))
    .catch((e) => env.logger.error('failed to send foo', e));

C

cgOP•11/23/23, 9:37 PM

well how about that

C

cgOP•11/23/23, 9:37 PM

some send and some don't

C

cgOP•11/23/23, 9:38 PM

but many don't

C

cgOP•11/23/23, 9:41 PM

With the error:

 "Queue send failed: Service Temporarily Unavailable"

 "Queue send failed: Service Temporarily Unavailable"

Ccg but many don't

M

Matt Silverlock•11/23/23, 9:50 PM

OK, let us continue to investigate.

C

cgOP•11/23/23, 10:05 PM

I have removed taht foo code

C

cgOP•11/23/23, 10:05 PM

I'm going to have to leave now, thanksgiving stuff.

C

cgOP•11/23/23, 10:05 PM

I'll check back in

C

cgOP•11/23/23, 10:06 PM

If you have any updates Matt, I have a support ticket

D

dattaproffs•11/23/23, 11:30 PM

Hi, i don't see anything about it in the docs but i thought i could ask anyway.
Is there a way to delay messages in the queue? i wan't to add them to the queue but consume them in an hour.

C

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

@Matt any news/ updates?

Ccg @Matt any news/ updates?

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