I will just pour through the itty-code and figure out what I can, I will post back what I figure ou

I will just pour through the itty-code and figure out what I can, I will post back what I figure out in case anyone else runs into this.

MMatt Silverlock Some updates for @Queues users: - We're working on landing HTTP Pull, which wil...

switz•11/22/23, 7:04 PM

huge!! very much waiting for this feature, will get me off SQS entirely.

switz•11/22/23, 7:04 PM

and subsequently off s3

quambo•11/22/23, 10:23 PM

Does unstable_devunstable_dev work with queues? I've got 2 workers locally, one setup as consumer, one as producer. When the producer sends into the queue, the consumer does not react. Is this supposed to work in local dev?

Qquambo Does `unstable_dev` work with queues? I've got 2 workers locally, one setup as c...

quambo•11/22/23, 10:35 PM

Doesnt work as it seems, from Docs:
Wrangler does not yet support running separate producer and consumer Workers bound to the same Queue locally. To develop locally with Queues, you can temporarily put your consumer’s queue() handler in the same Worker as your producer, so the same Worker acts as both a producer and consumer.

PPaul Shriner I will just pour through the itty-code and figure out what I can, I will post b...

Paul ShrinerOP•11/23/23, 5:16 AM

OK, got it. And the solution was itty-durable related, so...

If you are using the consumer of the queue, and are trying to pull a DO without using itty-router. This is how I solved it.

I used the proxyDurable class which isn't well documented as of writing this. The key though was not to just pass in the binding, but to also pass in the name of the binding in the options argument.

My binding was called "USER", so I would access it by env.USER which is the primary argument in proxyDurable.

let user = proxyDurable(this.env.USER, { name: "USER", parse: true })

let user = proxyDurable(this.env.USER, { name: "USER", parse: true })

then allows me to call the specific DO that I am looking for.

user.get(uid)

user.get(uid)

cg•11/23/23, 6:13 PM

Are Queues down at the moment, I'm seeing

Service Temporarily Unavailable

Service Temporarily Unavailable

cg•11/23/23, 6:13 PM

Is there a Queues status page I can check?

Ccg Are Queues down at the moment, I'm seeing `Service Temporarily Unavailable`

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

Can you expand how you’re getting this error? I don’t see any service wide issues.

cg•11/23/23, 7:25 PM

From within a worker, I'm invoking

sendBatch

sendBatch

. This error is being logged to my 3rd party logging infra

cg•11/23/23, 7:26 PM

The pattern here is that the consumer is calling sendBatch to process the next "page" of data

cg•11/23/23, 7:26 PM

as to avoid the 1000 sub request limit

TTom Hi, I'm getting an error on a dead-letter queue consumer that I am unable to ex...

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

We suspect this is a bug between the content-type option & DLQs.

Ccg From within a worker, I'm invoking `sendBatch` . This error is being logged to m...

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

Can you share your account ID? (it’s safe to share here)

cg•11/23/23, 7:29 PM

Sure, give me a sec

cg•11/23/23, 7:30 PM

Account ID:

10331f908d1d8a668c229d6e85654c7f

10331f908d1d8a668c229d6e85654c7f

Ccg Sure, give me a sec

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

I also assume you don’t have a huge backlog?

cg•11/23/23, 7:30 PM

I cleared out the backlog last night

cg•11/23/23, 7:30 PM

Was sitting at like 250k that were just sitting there, so I uh programatically ack'd them all

cg•11/23/23, 7:38 PM

I've submitted a ticket and also reached out to our contact Dylan

Ccg Was sitting at like 250k that were just sitting there, so I uh programatically a...

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

Got it. I don't see anything obvious here from a quick look at the metrics.

cg•11/23/23, 7:40 PM

As a FYI, I do have the max_message_batch size as 1, because I've found that if it's too high and one worker handles like 10 messages that each make 100 requests, then the queue just dies (somewhat random and silently) due to the max req limit

cg•11/23/23, 7:40 PM

I'm not sure that has anything to do with this however

Ccg As a FYI, I do have the max_message_batch size as 1, because I've found that if ...

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

which queue ID is it? You have two.

8a6de0

8a6de0

89a515

89a515

cg•11/23/23, 7:44 PM

Uh, I can't actually see the Q Id in the dashboard

cg•11/23/23, 7:44 PM

let me look harder

cg•11/23/23, 7:44 PM

it's

89a515eb79...

89a515eb79...

Ccg it's `89a515eb79...`

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

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)

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

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

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

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.

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

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

oh excuse me, I might be wrong

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

Thanks, I'll take a look at that

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

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

sendBatch

sendBatch

counts as one) . An invocation is each call to

queue

queue

fetch

fetch

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

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

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

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

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

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

I've also seen the error

Queue send failed: Service Temporarily Unavailable

Queue send failed: Service Temporarily Unavailable

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

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

Correct, so we can rule that out.

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

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

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

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

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

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

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

Heck, just

send(“foo”)

send(“foo”)

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

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

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

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

I will just pour through the itty-code and figure out what I can, I will post back what I figure ou

Similar Threads