Explicit message pulling with `consume()`

[albnnc]

Proposed change

Hello!

Thanks in advance for reading this.

Recently I've asked a question and received an answer to it. The question was about batched message consuming: I've wanted to receive a batch of JetStream messages and start processing them before reader expiration while not receiving any new messages. The answer purposes using fetch() and breaking out from consumption loop if there're no pending messages. This is logical indeed, but it seems to fail in certain scenarios.

I've transformed code from answer a little bit in order to show the problem:

(1) Example with fetch()

import {
  AckPolicy,
  jetstreamManager,
  type JsMsg,
} from "jsr:@nats-io/jetstream@3.0.0-38";
import {
  connect,
  delay,
  nanos,
  nuid,
} from "jsr:@nats-io/transport-deno@3.0.0-22";
import { chunk } from "jsr:@std/collections@1.0.10/chunk";

const nc = await connect({
  servers: [
    // "demo.nats.io",
    "nats://localhost:32000",
  ],
});
const name = nuid.next();

const jsm = await jetstreamManager(nc);
const js = jsm.jetstream();

await jsm.streams.add({
  name,
  subjects: [`${name}.>`],
  max_age: nanos(20_000),
});

await jsm.consumers.add(
  name,
  {
    name,
    ack_policy: AckPolicy.Explicit,
    ack_wait: 10_000_000_000,
  },
);

const total = 10_000;
const publishChunkSize = 1_000;
const consumeChunkSize = 1_000;

const start = performance.now();
function log(...args: unknown[]) {
  const now = performance.now();
  const delta = Math.round((now - start) / 1_000) + " s";
  console.log(delta.padStart(6, " "), "|", ...args);
}

async function publish() {
  log("publishing start");
  for (
    const part of chunk(
      new Array(total).fill(undefined).map((_, i) => i),
      publishChunkSize,
    )
  ) {
    await Promise.all(part.map((v) => js.publish(`${name}.${v}`, `${v}`)));
    await delay(100);
  }
  log("publishing end");
}

async function consume() {
  log("consuming start");
  const c = await js.consumers.get(name, name);
  const buf: JsMsg[] = [];
  let count = 0;
  let redeliveryCount = 0;
  while (true) {
    const iter = await c.fetch({ max_messages: consumeChunkSize });
    for await (const m of iter) {
      count++;
      redeliveryCount += m.info.redelivered ? 1 : 0;
      buf.push(m);
      if (m.info.pending === 0) {
        break;
      }
    }
    log(
      `+${buf.length}`,
      `(${count} / ${total} messages, ${redeliveryCount} redeliveries)`,
    );
    buf.forEach((m) => {
      m.ack();
    });
    buf.length = 0;
    if (count >= total) {
      break;
    }
  }
  log("consuming end");
}

const consumePromise = consume();

await publish();

await consumePromise;

log("done");

await jsm.streams.delete(name);
await nc.close();

The most important changes are:

• We're subscribing to messages before their publish.
• We're publishing messages in uneven pace.

These changes correspond to real-world JetStream usage and lead to the following log:

Output

 0 s | consuming start
 0 s | publishing start
 0 s | +1000 (1000 / 10000 messages, 0 redeliveries)
 0 s | +31 (1031 / 10000 messages, 0 redeliveries)
 0 s | +284 (1315 / 10000 messages, 0 redeliveries)
 1 s | +13 (1328 / 10000 messages, 0 redeliveries)
 2 s | publishing end
10 s | +1000 (2328 / 10000 messages, 987 redeliveries)
11 s | +1000 (3328 / 10000 messages, 987 redeliveries)
11 s | +1000 (4328 / 10000 messages, 987 redeliveries)
11 s | +1000 (5328 / 10000 messages, 987 redeliveries)
11 s | +1000 (6328 / 10000 messages, 987 redeliveries)
11 s | +1000 (7328 / 10000 messages, 987 redeliveries)
11 s | +1000 (8328 / 10000 messages, 987 redeliveries)
11 s | +1000 (9328 / 10000 messages, 987 redeliveries)
11 s | +672 (10000 / 10000 messages, 987 redeliveries)
11 s | consuming end
11 s | done

What I think has happened:

• During first second message reading was working well.
• At the "1 s" mark we've received a message with m.info.pending === 0 and broken the loop to start processing messages.
• m.info.pending value was valid at the moment of message delivery, but it was wrong at the moment of loop break. This happened because we've been publishing messages in uneven pace.
• So, NATS server actually sent 987 new messages to old fetch-based consumer, but it was too late.
• We've processed a batch of messages at "1 s" mark, but the noted 987 messages were "lost" and waited for their redelivery for 10 seconds, because we've set ack_wait to that value.
• At the moment of redelivery all other messages have already been published, so we've got no any other delay until the end of 10k messages.

This problem gets much worse in real-world apps, where everything is delayed because of non-instant message processing. I'm unsure if it's a bug or protocol-related limitation.

Described above leads me to the following idea: it's better not to close reader and use consume-based reader. I think that this is the only possible way of consuming messages with OK-ish performance.

It looks like you recommend delaying inside for-await loop (1, 2), but I think that this a misdirection, since we'll have similar problem as described above: new messages will get picked up inside client while not being read in for-await. So, these messages (which are delayed by await process(...)) will get redelivered later to same or other consumer in case of queue group presence. The described can be seen in the following example (note different delays and limits):

(2) Example with consume()

import {
  AckPolicy,
  jetstreamManager,
  type JsMsg,
} from "jsr:@nats-io/jetstream@3.0.0-38";
import {
  connect,
  delay,
  nanos,
  nuid,
} from "jsr:@nats-io/transport-deno@3.0.0-22";
import { chunk } from "jsr:@std/collections@1.0.10/chunk";

const nc = await connect({
  servers: [
    // "demo.nats.io",
    "nats://localhost:32000",
  ],
});
const name = nuid.next();

const jsm = await jetstreamManager(nc);
const js = jsm.jetstream();

await jsm.streams.add({
  name,
  subjects: [`${name}.>`],
  max_age: nanos(20_000),
});

await jsm.consumers.add(
  name,
  {
    name,
    ack_policy: AckPolicy.Explicit,
    ack_wait: 1_000_000_000,
  },
);

const total = 10_000;
const publishChunkSize = 2_000;
const consumeChunkSize = 1_000;

const start = performance.now();
function log(...args: unknown[]) {
  const now = performance.now();
  const delta = Math.round((now - start) / 1_000) + " s";
  console.log(delta.padStart(6, " "), "|", ...args);
}

async function publish() {
  log("publishing start");
  for (
    const part of chunk(
      new Array(total).fill(undefined).map((_, i) => i),
      publishChunkSize,
    )
  ) {
    await Promise.all(part.map((v) => js.publish(`${name}.${v}`, `${v}`)));
    await delay(100);
  }
  log("publishing end");
}

async function consume() {
  log("consuming start");
  const c = await js.consumers.get(name, name);
  const buf: JsMsg[] = [];
  const process = async () => {
    log(
      `+${buf.length}`,
      `(${count} / ${total} messages, ${redeliveryCount} redeliveries)`,
    );
    await delay(2_000);
    buf.forEach((m) => {
      m.ack();
    });
    buf.length = 0;
  };
  let count = 0;
  let redeliveryCount = 0;
  const iter = await c.consume({ max_messages: consumeChunkSize });
  for await (const m of iter) {
    count++;
    redeliveryCount += m.info.redelivered ? 1 : 0;
    buf.push(m);
    if (
      m.info.pending === 0 ||
      buf.length >= consumeChunkSize
      // Actually, we can't depend on just `m.info.pending`
      // and `buf.length` here, since we can get `m.info.pending > 0`
      // and no new messages later in case of multiple queued consumers.
      // So, in real-world add I use a timer which checks buffer regularly
      // for inactivity and starts its processing if needed.
      // This is not the case here, however.
    ) {
      await process();
    }
    if (count >= total) {
      break;
    }
  }
  log("consuming end");
}

const consumePromise = consume();
await publish();
await consumePromise;

log("done");

await jsm.streams.delete(name);
await nc.close();

Output

   0 s | consuming start
   0 s | publishing start
   0 s | +1000 (1000 / 10000 messages, 0 redeliveries)
   2 s | publishing end
   2 s | +1000 (2000 / 10000 messages, 752 redeliveries)
   4 s | +1000 (3000 / 10000 messages, 1002 redeliveries)
   6 s | +1000 (4000 / 10000 messages, 1752 redeliveries)
   8 s | +1000 (5000 / 10000 messages, 2002 redeliveries)
  10 s | +1000 (6000 / 10000 messages, 2752 redeliveries)
  12 s | +1000 (7000 / 10000 messages, 3002 redeliveries)
  14 s | +1000 (8000 / 10000 messages, 3752 redeliveries)
  17 s | +1000 (9000 / 10000 messages, 4002 redeliveries)
  19 s | +1000 (10000 / 10000 messages, 4752 redeliveries)
  21 s | consuming end
  21 s | done

I think that picked up but unread messages need to be marked as being processed by sending ack(WIP) which I didn't find this library doing automatically. Indeed, it would be slow in some cases and should depend on certain consumer options.

Both examples above use docker.io/library/nats:2.10.26. However, while I was writing this issue and running isolated tests, I found out that the following example runs differently with different NATS versions:

(3) Example with consume() and different delays

import {
  AckPolicy,
  jetstreamManager,
  type JsMsg,
} from "jsr:@nats-io/jetstream@3.0.0-38";
import {
  connect,
  delay,
  nanos,
  nuid,
} from "jsr:@nats-io/transport-deno@3.0.0-22";
import { chunk } from "jsr:@std/collections@1.0.10/chunk";

const nc = await connect({
  servers: [
    // "demo.nats.io",
    "nats://localhost:32000",
  ],
});
const name = nuid.next();

const jsm = await jetstreamManager(nc);
const js = jsm.jetstream();

await jsm.streams.add({
  name,
  subjects: [`${name}.>`],
  max_age: nanos(20_000),
});

await jsm.consumers.add(
  name,
  {
    name,
    ack_policy: AckPolicy.Explicit,
    ack_wait: 10_000_000_000,
  },
);

const total = 10_000;
const publishChunkSize = 900;
const consumeChunkSize = 1_000;

const start = performance.now();
function log(...args: unknown[]) {
  const now = performance.now();
  const delta = Math.round((now - start) / 1_000) + " s";
  console.log(delta.padStart(6, " "), "|", ...args);
}

async function publish() {
  log("publishing start");
  for (
    const part of chunk(
      new Array(total).fill(undefined).map((_, i) => i),
      publishChunkSize,
    )
  ) {
    await Promise.all(part.map((v) => js.publish(`${name}.${v}`, `${v}`)));
    await delay(100);
  }
  log("publishing end");
}

async function consume() {
  log("consuming start");
  const c = await js.consumers.get(name, name);
  const buf: JsMsg[] = [];
  const process = async () => {
    log(
      `+${buf.length}`,
      `(${count} / ${total} messages, ${redeliveryCount} redeliveries)`,
    );
    await delay(2000);
    buf.forEach((m) => {
      m.ack();
    });
    buf.length = 0;
  };
  let count = 0;
  let redeliveryCount = 0;
  const iter = await c.consume({ max_messages: consumeChunkSize });
  for await (const m of iter) {
    count++;
    redeliveryCount += m.info.redelivered ? 1 : 0;
    buf.push(m);
    if (
      m.info.pending === 0 ||
      buf.length >= consumeChunkSize
      // Actually, we can't depend on just `m.info.pending`
      // and `buf.length` here, since we can get `m.info.pending > 0`
      // and no new messages later in case of multiple queued consumers.
      // So, in real-world add I use a timer which checks buffer regularly
      // for inactivity and starts its processing if needed.
      // This is not the case here, however.
    ) {
      await process();
    }
    if (count >= total) {
      break;
    }
  }
  log("consuming end");
}

const consumePromise = consume();
await publish();
await consumePromise;

log("done");

await jsm.streams.delete(name);
await nc.close();

Output with docker.io/library/nats:2.10.26

   0 s | consuming start
   0 s | publishing start
   0 s | +900 (900 / 10000 messages, 0 redeliveries)
   2 s | +10 (910 / 10000 messages, 0 redeliveries)
   2 s | publishing end
   4 s | +1 (911 / 10000 messages, 0 redeliveries)
   6 s | +1 (912 / 10000 messages, 0 redeliveries)
   8 s | +1 (913 / 10000 messages, 0 redeliveries)
  10 s | +1 (914 / 10000 messages, 0 redeliveries)
  12 s | +1 (915 / 10000 messages, 0 redeliveries)
  14 s | +1 (916 / 10000 messages, 0 redeliveries)
  16 s | +1 (917 / 10000 messages, 0 redeliveries)
  18 s | +1 (918 / 10000 messages, 0 redeliveries)
  20 s | +1 (919 / 10000 messages, 0 redeliveries)
  22 s | +1 (920 / 10000 messages, 0 redeliveries)
  24 s | +1 (921 / 10000 messages, 0 redeliveries)
  ...reading 1 message per second (why no redeliveries?)

Output with docker.io/library/nats:2.9.19

   0 s | consuming start
   0 s | publishing start
   0 s | +900 (900 / 10000 messages, 0 redeliveries)
   2 s | +23 (923 / 10000 messages, 0 redeliveries)
   2 s | publishing end
   4 s | +1000 (1923 / 10000 messages, 0 redeliveries)
   6 s | +1000 (2923 / 10000 messages, 0 redeliveries)
   8 s | +1000 (3923 / 10000 messages, 0 redeliveries)
  10 s | +1000 (4923 / 10000 messages, 0 redeliveries)
  12 s | +1000 (5923 / 10000 messages, 0 redeliveries)
  14 s | +1000 (6923 / 10000 messages, 0 redeliveries)
  16 s | +1000 (7923 / 10000 messages, 0 redeliveries)
  18 s | +1000 (8923 / 10000 messages, 0 redeliveries)
  20 s | +1000 (9923 / 10000 messages, 0 redeliveries)
  22 s | +77 (10000 / 10000 messages, 0 redeliveries)
  24 s | consuming end
  24 s | done

It's hard for me to understand what's happening here.

I think that the good solution would be:

• Read messages in a non-blocking manner, without await process(...) inside consuming loop.
• Separate messages in two pools, the one which is being processed and the one which is being collected.
• Depend on the max_messages and request new messages from NATS server depending on how many messages have been processed. I propose this being done by:
  • Explicitly stating in documentation that consume({ threshold_messages: -1, threshold_bytes: -1 }) is not a hack. Or providing another way to disable automatic message requesting.
  • Providing control over message requesting via exposing something like reader.pull(...), which already exists but not presented in interface.

Use case

With the proposed changes it would be possible to consume messages like that:

(4) Example with consume() and proposed changes

import {
  AckPolicy,
  jetstreamManager,
  type JsMsg,
} from "jsr:@nats-io/jetstream@3.0.0-38";
import {
  connect,
  delay,
  nanos,
  nuid,
} from "jsr:@nats-io/transport-deno@3.0.0-22";
import { chunk } from "jsr:@std/collections@1.0.10/chunk";

const nc = await connect({
  servers: [
    // "demo.nats.io",
    "nats://localhost:32000",
  ],
});
const name = nuid.next();

const jsm = await jetstreamManager(nc);
const js = jsm.jetstream();

await jsm.streams.add({
  name,
  subjects: [`${name}.>`],
  max_age: nanos(20_000),
});

await jsm.consumers.add(
  name,
  {
    name,
    ack_policy: AckPolicy.Explicit,
    ack_wait: 10_000_000_000,
  },
);

const total = 10_000;
const publishChunkSize = 900;
const consumeChunkSize = 1_000;

const start = performance.now();
function log(...args: unknown[]) {
  const now = performance.now();
  const delta = Math.round((now - start) / 1_000) + " s";
  console.log(delta.padStart(6, " "), "|", ...args);
}

async function publish() {
  log("publishing start");
  for (
    const part of chunk(
      new Array(total).fill(undefined).map((_, i) => i),
      publishChunkSize,
    )
  ) {
    await Promise.all(part.map((v) => js.publish(`${name}.${v}`, `${v}`)));
    await delay(100);
  }
  log("publishing end");
}

async function consume() {
  log("consuming start");
  const c = await js.consumers.get(name, name);
  let processingBuf: JsMsg[] | undefined;
  let collectingBuf: JsMsg[] = [];
  const process = async () => {
    if (processingBuf) {
      throw new Error("Already processing");
    }
    processingBuf = collectingBuf;
    collectingBuf = [];
    log(
      `+${processingBuf.length}`,
      `(${count} / ${total} messages, ${redeliveryCount} redeliveries)`,
    );
    pull();
    await delay(2000);
    processingBuf.forEach((m) => {
      m.ack();
    });
    processingBuf = undefined;
    check();
  };
  let count = 0;
  let redeliveryCount = 0;
  const iter = await c.consume({
    max_messages: consumeChunkSize,
    threshold_messages: -1,
    threshold_bytes: -1,
  });
  const pull = () => {
    // deno-lint-ignore no-explicit-any
    const options = (iter as any).pullOptions();
    options.n = consumeChunkSize - iter.getPending();
    if (options.n < 1) {
      return;
    }
    // deno-lint-ignore no-explicit-any
    (iter as any).pull(options);
  };
  const check = () => {
    if (
      !processingBuf && (
        collectingBuf.at(-1)?.info.pending === 0 ||
        collectingBuf.length >= consumeChunkSize
      )
      // Actually, we can't depend on just `m.info.pending`
      // and `buf.length` here, since we can get `m.info.pending > 0`
      // and no new messages later in case of multiple queued consumers.
      // So, in real-world add I use a timer which checks buffer regularly
      // for inactivity and starts its processing if needed.
      // This is not the case here, however.
    ) {
      process();
    }
  };
  for await (const m of iter) {
    count++;
    redeliveryCount += m.info.redelivered ? 1 : 0;
    collectingBuf.push(m);
    check();
    if (count >= total) {
      break;
    }
  }
  log("consuming end");
}

const consumePromise = consume();
await publish();
await consumePromise;

log("done");

await jsm.streams.delete(name);
await nc.close();

Output

   0 s | consuming start
   0 s | publishing start
   0 s | +900 (900 / 10000 messages, 0 redeliveries)
   1 s | +1000 (1900 / 10000 messages, 0 redeliveries)
   2 s | +1000 (2900 / 10000 messages, 0 redeliveries)
   2 s | publishing end
   3 s | +1000 (3900 / 10000 messages, 0 redeliveries)
   4 s | +1000 (4900 / 10000 messages, 0 redeliveries)
   5 s | +1000 (5900 / 10000 messages, 0 redeliveries)
   6 s | +1000 (6900 / 10000 messages, 0 redeliveries)
   7 s | +1000 (7900 / 10000 messages, 0 redeliveries)
   8 s | +1000 (8900 / 10000 messages, 0 redeliveries)
   9 s | +1000 (9900 / 10000 messages, 0 redeliveries)
  10 s | +100 (10000 / 10000 messages, 0 redeliveries)
  10 s | consuming end
  10 s | done

Note that output contains expected timings and no redeliveries.

[albnnc]

Explicitly stating in documentation that consume({ threshold_messages: -1, threshold_bytes: -1 }) is not a hack. Or providing another way to disable automatic message requesting.

It's worth to say that even disabling message requesting with consume({ threshold_messages: -1, threshold_bytes: -1 }) won't be enough probably, since client will re-request messages after initial request expiration (won't it?), which is going to conflict with manual messages requesting.

So, it would be nice to have a separate option to disable all automatic message pulls.

[aricart]

Firstly - the issue here is the break while consuming or fetching, and then blocking of processing.
The first rule is that the size of a batch (in either consume or fetch) cannot exceed the number of messages that you can safely process within the ack window UNLESS you are a single consumer. In that case, so long as you ack the message, it will be OK.

If you ask for a batch of 10, and the server has 10 messages, the clock on the acks starts running before the for loop is presented with those messages. So if you have an ack window of 10s, and each message takes 2 seconds to process, by the time you are done with the 4th message you could run into trouble as it is possible that message 5 will not make it in time. For sure, messages 6-10 are expired and candidates for a re-delivery.

See:

import { connect, nuid, nanos, delay } from "@nats-io/transport-deno";
import {jetstreamManager, jetstream, AckPolicy} from "@nats-io/jetstream";


const nc = await connect({servers: "demo.nats.io"});
console.log(`connected to ${nc.getServer()}`)
const jsm = await jetstreamManager(nc);

const stream = nuid.next();
await jsm.streams.add({name: stream, subjects: [stream]});
await jsm.consumers.add(stream, {name: "c", ack_policy: AckPolicy.Explicit, ack_wait: nanos(10_000)});

const js = jsm.jetstream();
const buf = [];
for(let i=0; i < 100; i++) {
  buf.push(js.publish(stream, `${i}`));
}

await Promise.all(buf);

const c = await js.consumers.get(stream, "c");
console.log(await c.info());

const iter = await c.fetch({max_messages: 10});
for await(const m of iter) {
  console.log(m.seq);
  await delay(2_000);
  m.ack();
}

await nc.flush();
console.log(await c.info());

await jsm.streams.delete(stream);
await nc.close();

Which prints:

deno run -A main.ts
connected to demo.nats.io:4222
{
  type: "io.nats.jetstream.api.v1.consumer_info_response",
  stream_name: "6SNQJAH0ITUJ1W3TE7ZYW8",
  name: "c",
  created: "2025-03-12T22:33:05.673240651Z",
  config: {
    name: "c",
    deliver_policy: "all",
    ack_policy: "explicit",
    ack_wait: 10000000000,
    max_deliver: -1,
    replay_policy: "instant",
    max_waiting: 512,
    max_ack_pending: 1000,
    inactive_threshold: 5000000000,
    num_replicas: 0
  },
  delivered: { consumer_seq: 0, stream_seq: 0 },
  ack_floor: { consumer_seq: 0, stream_seq: 0 },
  num_ack_pending: 0,
  num_redelivered: 0,
  num_waiting: 0,
  num_pending: 100,
  ts: "2025-03-12T22:33:05.884004538Z"
}
1
2
3
4
5
6
7
8
9
10
{
  type: "io.nats.jetstream.api.v1.consumer_info_response",
  stream_name: "6SNQJAH0ITUJ1W3TE7ZYW8",
  name: "c",
  created: "2025-03-12T22:33:05.673240651Z",
  config: {
    name: "c",
    deliver_policy: "all",
    ack_policy: "explicit",
    ack_wait: 10000000000,
    max_deliver: -1,
    replay_policy: "instant",
    max_waiting: 512,
    max_ack_pending: 1000,
    inactive_threshold: 5000000000,
    num_replicas: 0
  },
  delivered: {
    consumer_seq: 10,
    stream_seq: 10,
    last_active: "2025-03-12T22:33:06.023785518Z"
  },
  ack_floor: {
    consumer_seq: 10,
    stream_seq: 10,
    last_active: "2025-03-12T22:33:26.107011376Z"
  },
  num_ack_pending: 0,
  num_redelivered: 0,
  num_waiting: 0,
  num_pending: 90,
  ts: "2025-03-12T22:33:26.200870922Z"
}

Note that above there's a single worker. Note that the consumer info printed doesn't show any pending acks (even though they were late), and there are no redeliveries because the client didn't fetch more messages from the consumer.

The minute you have more than worker on the same consumer, this doesn't work. Because messages 6-10 are candidates for redelivery, and are sent to the other worker (even if the first worker also saw them).

What is the correct strategy?

The easy one is to fetch as many messages as you process giving some time for the ack to get back to the server. So in the above scenario, you can only really process 4, so ask for 4.

If you want to scale without having to perform such calculations, instead, you should just have each worker use next() and not try to schedule the work, if you add 100 processes, each doing a next() with expires of 60s, you possibly will process all 100 messages within 2 seconds or so.

[aricart]

BTW, the whole idea of consume is that it keeps a buffer ready, this is not the issue and increase the utilization of the client since messages will be ready for processing. In your case you should never use consume(), and instead should only fetch() or next().

[aricart]

This is also contrived as you'll be competing for CPU (you can use webworkers with their own connection sharing the consumer name to unblock):

await jsm.consumers.add(stream, {name: "cc", ack_policy: AckPolicy.Explicit, ack_wait: nanos(10_000)});
const cc = await js.consumers.get(stream, "cc");


async function work(i: number): Promise<void> {
  const m = await cc.next();
  console.log(`worker ${i}`, m.seq);
  await delay(2_000);
  m.ack();
}

const start = Date.now();
buf.length = 0;
for(let i=0; i < 100; i++) {
  buf.push(work(i));
}

await Promise.all(buf);
console.log(`processing time: ${Date.now() - start}`)
console.log(await cc.info());

worker 0 1
worker 1 2
worker 2 3
worker 3 4
worker 4 5
worker 5 6
worker 6 7
worker 7 8
worker 8 9
worker 9 10
worker 10 11
worker 11 12
worker 12 13
worker 13 14
worker 14 15
worker 15 16
worker 16 17
worker 17 18
worker 18 19
worker 19 20
worker 20 21
worker 21 22
worker 22 23
worker 23 24
worker 24 25
worker 25 26
worker 26 27
worker 27 28
worker 28 29
worker 29 30
worker 30 31
worker 31 32
worker 32 33
worker 33 34
worker 34 35
worker 35 36
worker 36 37
worker 37 38
worker 38 39
worker 39 40
worker 40 41
worker 41 42
worker 42 43
worker 43 44
worker 44 45
worker 45 46
worker 46 47
worker 47 48
worker 48 49
worker 49 50
worker 50 51
worker 51 52
worker 52 53
worker 53 54
worker 54 55
worker 55 56
worker 56 57
worker 57 58
worker 58 59
worker 59 60
worker 60 61
worker 61 62
worker 62 63
worker 63 64
worker 64 65
worker 65 66
worker 66 67
worker 67 68
worker 68 69
worker 69 70
worker 70 71
worker 71 72
worker 72 73
worker 73 74
worker 74 75
worker 75 76
worker 76 77
worker 77 78
worker 78 79
worker 79 80
worker 80 81
worker 81 82
worker 82 83
worker 83 84
worker 84 85
worker 85 86
worker 86 87
worker 87 88
worker 88 89
worker 89 90
worker 90 91
worker 91 92
worker 92 93
worker 93 94
worker 94 95
worker 95 96
worker 96 97
worker 97 98
worker 98 99
worker 99 100
processing time: 2120
{
  type: "io.nats.jetstream.api.v1.consumer_info_response",
  stream_name: "5747PWI884ZQKE7KZWC7JL",
  name: "cc",
  created: "2025-03-12T23:06:58.851462952Z",
  config: {
    name: "cc",
    deliver_policy: "all",
    ack_policy: "explicit",
    ack_wait: 10000000000,
    max_deliver: -1,
    replay_policy: "instant",
    max_waiting: 512,
    max_ack_pending: 1000,
    inactive_threshold: 5000000000,
    num_replicas: 0
  },
  delivered: {
    consumer_seq: 100,
    stream_seq: 100,
    last_active: "2025-03-12T23:06:59.016977618Z"
  },
  ack_floor: {
    consumer_seq: 100,
    stream_seq: 100,
    last_active: "2025-03-12T23:07:01.121334962Z"
  },
  num_ack_pending: 0,
  num_redelivered: 0,
  num_waiting: 0,
  num_pending: 0,
  ts: "2025-03-12T23:07:01.121379629Z"
}

[albnnc]

Firstly - the issue here is the break while consuming or fetching, and then blocking of processing. The first rule is that the size of a batch (in either consume or fetch) cannot exceed the number of messages that you can safely process within the ack window UNLESS you are a single consumer. In that case, so long as you ack the message, it will be OK.

I mostly agree with this, but I also think that trying to guess the size of a batch which is feasible for app to process in time is too tricky and looks like magical hackity-hack. It looks like +WPI ack type was designed specifically to handle this "problem". Since you're also writing about it, I guess there's no misunderstanding.

If you ask for a batch of 10, and the server has 10 messages, the clock on the acks starts running before the for loop is presented with those messages. So if you have an ack window of 10s, and each message takes 2 seconds to process, by the time you are done with the 4th message you could run into trouble as it is possible that message 5 will not make it in time. For sure, messages 6-10 are expired and candidates for a re-delivery.

Note that above there's a single worker. Note that the consumer info printed doesn't show any pending acks (even though they were late), and there are no redeliveries because the client didn't fetch more messages from the consumer.

The minute you have more than worker on the same consumer, this doesn't work. Because messages 6-10 are candidates for redelivery, and are sent to the other worker (even if the first worker also saw them).

While I do understand your example and explanation, I can't apply it to my example (3) with nats:2.10.26. Thanks for example anyways.

If you want to scale without having to perform such calculations, instead, you should just have each worker use next() and not try to schedule the work, if you add 100 processes, each doing a next() with expires of 60s, you possibly will process all 100 messages within 2 seconds or so.

Using next() is extremely slow. Also using 100 processes to handle 100 messages in 2 seconds seems very unreasonable. In my team we're handling thousands of messages per second with far less process count.

BTW, the whole idea of consume is that it keeps a buffer ready, this is not the issue and increase the utilization of the client since messages will be ready for processing. In your case you should never use consume(), and instead should only fetch() or next().

I actually don't know why I was going in circles with all these things. It seems that I failed with fetch() usage when I was writing original question in discussions and didn't think of it now when I stopped breaking consumer loop. Here's what using fetch() with separate message pools looks like:

(5) Example with fetch() and split buffers

import {
  AckPolicy,
  jetstreamManager,
  type JsMsg,
} from "jsr:@nats-io/jetstream@3.0.0-38";
import {
  connect,
  delay,
  nanos,
  nuid,
} from "jsr:@nats-io/transport-deno@3.0.0-22";
import { chunk } from "jsr:@std/collections@1.0.10/chunk";

const nc = await connect({
  servers: [
    // "demo.nats.io",
    "nats://localhost:32000",
  ],
});
const name = nuid.next();

const jsm = await jetstreamManager(nc);
const js = jsm.jetstream();

await jsm.streams.add({
  name,
  subjects: [`${name}.>`],
  max_age: nanos(20_000),
});

await jsm.consumers.add(
  name,
  {
    name,
    ack_policy: AckPolicy.Explicit,
    ack_wait: 1_000_000_000,
  },
);

const c = await js.consumers.get(name, name);

const total = 10_000;
const publishChunkSize = 900;
const consumeChunkSize = 1_000;

const start = performance.now();
function log(...args: unknown[]) {
  const now = performance.now();
  const delta = Math.round((now - start) / 1_000) + " s";
  console.log(delta.padStart(6, " "), "|", ...args);
}

async function publish() {
  log("publishing start");
  for (
    const part of chunk(
      new Array(total).fill(undefined).map((_, i) => i),
      publishChunkSize,
    )
  ) {
    await Promise.all(part.map((v) => js.publish(`${name}.${v}`, `${v}`)));
    await delay(100);
  }
  log("publishing end");
}

async function consume() {
  log("consuming start");
  let processingBuf: JsMsg[] | undefined;
  let collectingBuf: JsMsg[] = [];
  let done = false;
  (async () => {
    while (!done) {
      processingBuf?.forEach((v) => v.working());
      collectingBuf.forEach((v) => v.working());
      await delay(500);
    }
  })();
  const process = async () => {
    if (processingBuf) {
      throw new Error("Already processing");
    }
    processingBuf = collectingBuf;
    collectingBuf = [];
    log(
      `+${processingBuf.length}`,
      `(${count} / ${total} messages, ${redeliveryCount} redeliveries)`,
    );
    await delay(2000);
    processingBuf.forEach((m) => {
      m.ack();
    });
    processingBuf = undefined;
    check();
  };
  let count = 0;
  let redeliveryCount = 0;
  const check = () => {
    if (
      !processingBuf && (
        collectingBuf.at(-1)?.info.pending === 0 ||
        collectingBuf.length >= consumeChunkSize
      )
      // Actually, we can't depend on just `m.info.pending`
      // and `buf.length` here, since we can get `m.info.pending > 0`
      // and no new messages later in case of multiple queued consumers.
      // So, in real-world add I use a timer which checks buffer regularly
      // for inactivity and starts its processing if needed.
      // This is not the case here, however.
    ) {
      process();
    }
  };
  while (!done) {
    const requestCount = consumeChunkSize - collectingBuf.length;
    const reader = await c.fetch({ max_messages: requestCount });
    for await (const m of reader) {
      count++;
      redeliveryCount += m.info.redelivered ? 1 : 0;
      collectingBuf.push(m);
      check();
      if (count >= total) {
        done = true;
        break;
      }
    }
  }
  log("consuming end");
}

const consumePromise = consume();
await publish();
await consumePromise;

log("done");

await jsm.streams.delete(name);
await nc.close();

Output

   0 s | consuming start
   0 s | publishing start
   0 s | +651 (651 / 10000 messages, 0 redeliveries)
   2 s | +349 (1000 / 10000 messages, 0 redeliveries)
   2 s | publishing end
   4 s | +1000 (2000 / 10000 messages, 0 redeliveries)
   6 s | +1000 (3000 / 10000 messages, 0 redeliveries)
   8 s | +1000 (4000 / 10000 messages, 0 redeliveries)
  10 s | +1000 (5000 / 10000 messages, 0 redeliveries)
  12 s | +1000 (6000 / 10000 messages, 0 redeliveries)
  14 s | +1000 (7000 / 10000 messages, 0 redeliveries)
  16 s | +1000 (8000 / 10000 messages, 0 redeliveries)
  18 s | +1000 (9000 / 10000 messages, 0 redeliveries)
  20 s | +1000 (10000 / 10000 messages, 0 redeliveries)
  20 s | consuming end
  20 s | done

Yes, it looks OK. All the benchmarks I have of message consumption with consume() + manual pulling and fetch() resulted in roughly the same timings.

The only thing about fetch() which is bothering me is that with consume() we can request additional messages before previous request expiration or fulfillment and thus preventing consumer idle time. Shouldn't consume() be theoretically faster then? And yes, we could do multiple fetch()-based requests in parallel, but I felt that it can be slower in the end.

This is also contrived as you'll be competing for CPU (you can use webworkers with their own connection sharing the consumer name to unblock)

Yes, I've already encountered it and tried to investigate performance bottlenecks. I think that most of the time is taken by protocol decoding and periodic garbage collecting of RequestOne variables which are constructed during message publish.

Also, is it expected for publishing using nats.js under Deno to be 2-3x slower than js.PublishAsync under Go?

Haven't you thought about using WASM/WASI to handle connection and protocol decoding/encoding? For example, AFAIK Pulsar Node client uses FFI (which seems unhandy in the world of different runtimes to be honest).

Thanks again, feel free to close this issue.

[aricart]

At some point the implementation needs to figure out the runtime/performance tradeoff - and in some cases perhaps JavaScript is not the right language for it. For the JavaScript client the goal is to support many runtimes.

There's one setting that currently the new consumers don't implement which terminates a batch request with whatever the server can give you. Thinking that this is possibly a setting that you could use. The issue with it is that your client will be slamming the server because as soon as you fetch, you'll do it again, and the API call won't throttle you.

That would allow you to do:

fetch({max_messages: 1000})

the server will give you up to 1000 messages. But if it gives you less, it doesn't have what you asked.

Let me dig into it.

[albnnc]

For the JavaScript client the goal is to support many runtimes.

Agree, that's why I think that FFI is a bad choice for a JavaScript-related library. It was easy to setup very basic nats.go-based client with FFI, but I don't think it's worth of creating complete library bindings. WASM looks way better, but I failed to spin up nats.go with WASI.

There's one setting that currently the new consumers don't implement which terminates a batch request with whatever the server can give you. Thinking that this is possibly a setting that you could use. The issue with it is that your client will be slamming the server because as soon as you fetch, you'll do it again, and the API call won't throttle you.

You're probably talking about no_wait. If so, I agree that it will result in either overloaded server or magic delays between fetch() calls. Overloaded server is bad, and those delays will be slower than an open message request for sure, so I'm pretty sure that it won't work well.

Actually, right now I'm testing a project with connector code refactored to use logic which is very similar to example (5). So far it works good, performance is OK, and I don't see any major drawbacks. Thus, I really hope that after adding no_wait support fetch() will still be able to create basic requests.