Seems fine to me

O

Old Man Maker•10/23/23, 11:27 AM

Yeah, I'm a little stumped. Definitely can't read or write to the kv and the dashboard logging is show me all the console logs up to the try/catch error and then no information from the error, so I'm kinda blind here

K

kianOP•10/23/23, 11:29 AM

What does your Wrangler.toml look like?

O

Old Man Maker•10/23/23, 11:31 AM

Normal:

[[kv_namespaces]]
binding = "AIS_ALL_TRANSACTIONS"
id = "xxxxx" <=I have the correct Id in here

[[queues.consumers]]
queue = "my-q" <=Correct queue here as well



# Optional: Configure batching and retries: https://developers.cloudflare.com/queues/learning/batching-retries/
# max_batch_size = 10 //The maximum number of messages to deliver before the timeout
# max_batch_timeout = 30 //default is 5 seconds
# max_retries = 10 //default to 3

# all dead transactions go to the following dead letter queue
dead_letter_queue = "transaction-dead-q"

[[kv_namespaces]]
binding = "AIS_ALL_TRANSACTIONS"
id = "xxxxx" <=I have the correct Id in here

[[queues.consumers]]
queue = "my-q" <=Correct queue here as well



# Optional: Configure batching and retries: https://developers.cloudflare.com/queues/learning/batching-retries/
# max_batch_size = 10 //The maximum number of messages to deliver before the timeout
# max_batch_timeout = 30 //default is 5 seconds
# max_retries = 10 //default to 3

# all dead transactions go to the following dead letter queue
dead_letter_queue = "transaction-dead-q"

[[kv_namespaces]]
binding = "AIS_ALL_TRANSACTIONS"
id = "xxxxx" <=I have the correct Id in here

[[queues.consumers]]
queue = "my-q" <=Correct queue here as well



# Optional: Configure batching and retries: https://developers.cloudflare.com/queues/learning/batching-retries/
# max_batch_size = 10 //The maximum number of messages to deliver before the timeout
# max_batch_timeout = 30 //default is 5 seconds
# max_retries = 10 //default to 3

# all dead transactions go to the following dead letter queue
dead_letter_queue = "transaction-dead-q"

[[kv_namespaces]]
binding = "AIS_ALL_TRANSACTIONS"
id = "xxxxx" <=I have the correct Id in here

[[queues.consumers]]
queue = "my-q" <=Correct queue here as well



# Optional: Configure batching and retries: https://developers.cloudflare.com/queues/learning/batching-retries/
# max_batch_size = 10 //The maximum number of messages to deliver before the timeout
# max_batch_timeout = 30 //default is 5 seconds
# max_retries = 10 //default to 3

# all dead transactions go to the following dead letter queue
dead_letter_queue = "transaction-dead-q"

Deploy shows its correctly attached

O

Old Man Maker•10/23/23, 11:34 AM

Is there anyway to get a catch error to show in the dashboard logs?

O

Old Man Maker•10/23/23, 2:50 PM

Moving this to KV channel as it is not queue specific

O

Old Man Maker•10/23/23, 3:31 PM

ok - I now understand what is going wrong.
In example code for queues, we often use a foreach loop to work through the batch. This has a nasty bug (the JS side of it) - I remembered coming across a similar issue a while back with async foreach loops in Node.
So, any async work needing to be done within the scope of the queue, you need to use a plain old for loop. This was hours and hours of debugging. I hope no-one else hits this, or at least people will know.

OOld Man Maker ok - I now understand what is going wrong. In example code for queues, we often ...

U

Unsmart•10/23/23, 3:41 PM

await does nothing on a foreach so this isnt really a bug its expected even if it seems weird: https://stackoverflow.com/questions/37576685/using-async-await-with-a-foreach-loop

U

Unsmart•10/23/23, 3:44 PM

I believe

Error Lens

Error Lens

Error Lens

Error Lens is the extension I have on vs code that makes it a bit easier to see these errors for me at least

OOld Man Maker ok - I now understand what is going wrong. In example code for queues, we often ...

J

John Spurlock•10/23/23, 3:45 PM

you are not the first to encounter this - maybe the cloudflare queue examples should explicitly mention avoiding

.forEach

.forEach

.forEach

.forEach

Debugging fetch in a consumer

UUnsmart I believe `Error Lens` is the extension I have on vs code that makes it a bit ea...

O

Old Man Maker•10/23/23, 4:31 PM

I'll take a look at that extension for sure. Thanks so much

S

Stefan•10/23/23, 10:30 PM

Hi, i'm currently trying to set up a queue between two different workers, one hosts the main application (the producer), and the other is responsible for firing off verification emails (the consumer), but what i'm not clear about and can't find documentation on, is that, assuming I have to have two wrangler.toml files (which my understanding is that i do), then how do i configure the queue to be exposed to both the consumer and the producer if they are in different files?

SStefan Hi, i'm currently trying to set up a queue between two different workers, one ho...

O

Old Man Maker•10/24/23, 7:13 AM

You add the producer details to the toml file of the producer workers (you can have multiple) - and you add the consumer details to the toml of the worker that is going to manage the queue.

S

Stefan•10/24/23, 8:47 AM

Okay so always pair the queue with the consumer, thank you!

�

😈 Donkey 💫•10/25/23, 5:29 AM

Hi everyone,
I have a question:

If:
Client http -> Producer worker -> queue -> consumer worker -> external http service api

How to make producer awaiting, then return response from "external service api" back to client

�

😈 Donkey 💫•10/25/23, 5:29 AM

is that possible?

K

kianOP•10/25/23, 5:43 AM

You can’t, no

K

kianOP•10/25/23, 5:43 AM

That wouldn’t be a use-case for Queues but rather Service Bindings

CCharlie What kind of message throughput are you atttempting to send here? What's the err...

A

arbfay•10/25/23, 12:17 PM

I found the problem. The issue was within a DO - I added a

this.env = env

this.env = env

this.env = env

this.env = env in the constructor to be able to call bindings from methods, then tried to send a last message to a queue right after calling

this.state.storage.deleteAll()

this.state.storage.deleteAll()

this.state.storage.deleteAll()

this.state.storage.deleteAll(), but it turns out calling

deleteAll()

deleteAll()

deleteAll()

deleteAll() also deletes the current memory and references of the object (not just storage), so the binding was lost. The error message is a bit confusing though - it made me believe the error was on CF's side (checked status page as well).

B

bret_pat•10/26/23, 10:22 AM

Are queue consumers going to be using the new "standard" worker pricing model in the future? Or at least, have the option to not be unbound?

Bbret_pat Are queue consumers going to be using the new "standard" worker pricing model in...

Y

Yongyi•10/26/23, 11:40 AM

i am also wondering the same thing

CCharlie @bret_pat @HeartYi consumers for Queues follow the same billing as any other wo...

Y

Yongyi•10/26/23, 4:20 PM

So they are billed based on cpu millisecond time not the wall time?

CCharlie https://blog.cloudflare.com/workers-pricing-scale-to-zero/

O

omar4289•10/26/23, 10:10 PM

Personally, I just think it got expensive for the unbundled mode (now 20$/1m seconds instead of 12.5$ and 0.3$ instead of 0.15$ per 1m requests). I know in the document we talk about the cpu time, instead of clock time, but to be honest my CPU time is usually very low, and thus the only impacting factor for me would be the number of requests, which unfortunately just doubled in price.

That would not change much for me, because Cloudflare is still a very generous platform and is pretty easy to use, I just don't like how this is promoted as only beneficial

Oomar4289 Personally, I just think it got expensive for the unbundled mode (now 20$/1m sec...

U

Unsmart•10/26/23, 10:17 PM

~100ms of wall time = + $0.15/million request

Also one unknown part of the unbound billing model is if you used 1 second of wall time and 1 second of CPU time you actually get billed 8 seconds wall time instead. (8x multiplier on CPU time)

So unless you had extremely fast requests it will probably be cheaper or around the same price.

UUnsmart ~100ms of wall time = + $0.15/million request Also one unknown part of the unbo...

O

omar4289•10/26/23, 10:21 PM

Lets see, but anyway, I still like the products Cloudflare is working ln. Really good job and I am open to pay any extra amount (if any) for the great services you provide at a very reasonable price

O

omar4289•10/26/23, 10:23 PM

Just one more question, since workers run on a 128MB machines, does it mean that the number of ms that we have is per 1GB of RAM, or is it per worker (128MB)?

U

Unsmart•10/26/23, 10:25 PM

The unbound billing model is based on 128MB (or 1/8th a GB) so 8 seconds of true wall time = 1 GB-sec for billing

O

omar4289•10/26/23, 10:30 PM

Ok, but right now, assuming I only run a loop in my program without I/O (very unlikely) and that takes exactly 1second to complete.

In the old model, I would pay 12.5$ for 8 million executions
In the new model, I would pay 20$ for 1 million executions.

I know the I/O work usually takes time, but it all depends on what you do, sometimes it will be cheaper, sometimes more expensive. Lets see how it goes

U

Unsmart•10/26/23, 10:37 PM

You can use this calculator and input your metrics to see the cost difference: https://calculator.imcf.me/

U

Unsmart•10/26/23, 10:38 PM

(The calculator doesnt take into account the CPU multiplier so you would need to make sure the wall time ms is at least 8x the cpu time ms)

M

Mad•10/28/23, 11:58 PM

I have a single queue marked both as a producer and a consumer queue, attached to the same worker. The worker takes input from the queue and produces output to the same queue until the task is done. What I observe is, the task runs for couple of iterations and stops running.

M

Mad•10/28/23, 11:58 PM

works fine when I run this locally

M

Mad•10/28/23, 11:59 PM

@Charlie B | KV, Queues, PubSub any idea what must be going wrong. Can I not use the same queue as a producer as well as consumer?

P

paper___•11/2/23, 2:53 AM

Hello, I'm having a problem regarding queues. I have a worker that streams and parses a big CSV file and puts one JSON object per row in a batch of 100 to a Queue. Locally, on dev, the flow works flawlessly, I can parse and enqueue a 250Mb CSV file with no problem and using low heap space. But in cloudflare, for some unknown reason the worker stops enqueuing at around 2600 or so objects with no error or warning whatsoever.

The code of the worker looks like this:

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

K

kianOP•11/2/23, 5:55 AM

Is that 2600 objects taking 30 seconds by any chance?

Kkian Is that 2600 objects taking 30 seconds by any chance?

P

paper___•11/2/23, 2:24 PM

Not really, it is 5 seconds top

Ppaper___Hello, I'm having a problem regarding queues. I have a worker that streams and p...

R

robmarscher•11/3/23, 4:31 PM

I had a similar issue that I think was due to the 1000 max subrequests per worker invocation. I think the queue send counts as a subrequest. https://developers.cloudflare.com/workers/platform/limits/#how-many-subrequests-can-i-make

That doesn't add up with 2600 objects though... that would only be 26 requests with a batch size of 100. 128 MB is the memory limit... but I assume that wouldn't be hit when you are streaming the body.

With sendBatch, there is a 256KB max size... maybe you are hitting that at a certain point in your data set and reducing the number of objects in sendBatch would fix?

A batch can contain up to 100 messages, though items are limited to 128 KB each, and the total size of the array cannot exceed 256 KB.

https://developers.cloudflare.com/queues/platform/javascript-apis/#queue

For the subrequest limit, I solved it by storing the uploaded csv on R2 and then using another queue to read the csv. That queue handler would track how many rows it processed and then once it reached 750 (arbitrary number to leave buffer under the 1000 limit) it would enqueue another task to read the csv - passing the number of rows to skip and the path to the csv on R2 in the queue message data. I also used a Durable Object to track the progress and push updates via websocket to the client that uploaded the csv.

Ppaper___Hello, I'm having a problem regarding queues. I have a worker that streams and p...

U

Unsmart•11/3/23, 4:35 PM

Try running the code without using ctx.waitUntil and see if you get any errors... its possible that waitUntil is swallowing the error

UUnsmart Try running the code without using ctx.waitUntil and see if you get any errors.....

M

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

This is very common - you will swalllow the exception inside

waitUntil

waitUntil

unless you are actively logging

C

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

Are Queues down at the moment?

Similar Threads

Similar Threads

Similar Threads