set up a queue consumer that's not scaling as much as I'd expect. The queue currently has ~500k mess

TTumbzilla set up a queue consumer that's not scaling as much as I'd expect. The queue curr...

E

Emilio•11/18/23, 1:11 AM

First thing I’d look at is how long the consumer script is taking to finish consuming a batch. For that you go look at the Worker metrics for the relevant script.

C

cg•11/18/23, 5:21 PM

Hey all! Is there a timeline for LogPush support in Queues?
It's quite hard to diagnose issues in Queues!

P

Paul Shriner•11/22/23, 4:38 PM

I cannot for the life of me figure out how to access Durable Objects from my consumer. I have been using itty-durable, but that has proven to be out of reach as I need the IRequest object. I would ideally like to use itty-durable, but at this point, just want it to work.

U

Unsmart•11/22/23, 4:42 PM

There is no request when using queue consumers... only a fetch has a request. You want to use the env which contains all of the bindings. The request also doesnt contain the bindings when in the fetch handler it is in the env parameter which is standard in all handlers.

P

Paul Shriner•11/22/23, 4:46 PM

Right, so when I try to do an env.DO.get("id"), it says that the id doesn't work. Hold on, let me get the exact error.

U

Unsmart•11/22/23, 4:47 PM

That is not a valid id... they look something like:

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

33ba9f42ca7b9084cf895328953209aae9da953df1a8d8f32f3832fe56

U

Unsmart•11/22/23, 4:47 PM

would also need to use

env.DO.idFromString()

env.DO.idFromString()

env.DO.idFromString()

env.DO.idFromString() to transform to a DO id

U

Unsmart•11/22/23, 4:48 PM

or env.DO.idFromName

P

Paul Shriner•11/22/23, 4:48 PM

true

P

Paul Shriner•11/22/23, 4:48 PM

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);
console.log(user)

P

Paul Shriner•11/22/23, 4:49 PM

sorry, I was shorthanding, and shouldn't have

U

Unsmart•11/22/23, 4:50 PM

I'd imagine you probably dont want

this.

this.

this.

this. should just be

env.

env.

env.

env.. But whats the error if thats what you have?

U

Unsmart•11/22/23, 4:53 PM

Example:

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

export default {
    async queue(batch: MessageBatch<any>, env: Bindings, ctx: ExecutionContext) {
        const stub = env.TestDO.get(env.TestDO.idFromName('1'))
        
        const response = await stub.fetch('https://fake-host/', {})
    }
}

P

Paul Shriner•11/22/23, 5:01 PM

When I call

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

let userid = this.env.USER.idFromName(uid)
let user = this.env.USER.get(userid);    
let brand = await user.GetBrand()

it says that the method GetBrand cannot be found that I would like to call for data access.

when I call

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

let userid = this.env.USER.idFromString(uid)
let user = this.env.USER.get(userid); 
let brand = await user.GetBrand()

This is the error code on the second:

Invalid Durable Object ID. Durable Object IDs must be 64 hex digits.

In both cases, the uid="GF5ZNCEJLV2X4UQ5FPMAPU6ZNRQ7OXOY3JPT6OOKUDRDEGKJX35TA2Y774"

P

Paul Shriner•11/22/23, 5:01 PM

Granted, in my non-consumer codes, I am using itty-durable and itty router, which it seems is not available here.

U

Unsmart•11/22/23, 5:11 PM

GetBrand is not a valid thing on durable objects you can only do fetch calls. An RPC like API is in the works though that would allow direct function calls.

P

Paul Shriner•11/22/23, 5:12 PM

Ahh, that makes sense. I can hit those methods with itty Durable, but not this way.

P

Paul Shriner•11/22/23, 5:16 PM

When I print out the user

console.log(JSON.stringify(user))

console.log(JSON.stringify(user))

console.log(JSON.stringify(user))

console.log(JSON.stringify(user))

It return an unpopulated object.

P

Paul Shriner•11/22/23, 5:19 PM

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

S

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

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

S

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

and subsequently off s3

Q

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

Does

unstable_dev

unstable_dev

unstable_dev

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

Q

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

P

Paul Shriner•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 })

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)

user.get(uid)

user.get(uid)

C

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

Are Queues down at the moment, I'm seeing

Service Temporarily Unavailable

Service Temporarily Unavailable

Service Temporarily Unavailable

Service Temporarily Unavailable

C

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`

M

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.

C

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

From within a worker, I'm invoking

sendBatch

sendBatch

sendBatch

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

C

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

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

C

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

M

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

M

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

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

C

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

Sure, give me a sec

C

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

Account ID:

10331f908d1d8a668c229d6e85654c7f

10331f908d1d8a668c229d6e85654c7f

10331f908d1d8a668c229d6e85654c7f

10331f908d1d8a668c229d6e85654c7f

Ccg Sure, give me a sec

M

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

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

C

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

I cleared out the backlog last night

C

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

C

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

M

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

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

C

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

C

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

M

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

which queue ID is it? You have two.

8a6de0

8a6de0

8a6de0

8a6de0 or

89a515

89a515

89a515

89a515

?

C

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

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

C

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

let me look harder

C

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

it's

89a515eb79...

89a515eb79...

89a515eb79...

89a515eb79...

Ccg it's `89a515eb79...`

M

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

Service Temporarily Unavailable

Service Temporarily Unavailable coming as an exception directly from

sendBatch

sendBatch

sendBatch

sendBatch

? (a code snippet would be helpful)

C

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

C

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

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

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

chunk

chunk

chunk

chunk here is

lodash

lodash

lodash

lodash's chunk function, to break messages into groups

set up a queue consumer that's not scaling as much as I'd expect. The queue currently has ~500k mess

Similar Threads

Similar Threads

Similar Threads