Uniqueness of messages consumed by pull consumer in new API

[whocaresk]

@aricart Hello and sorry that I tag you directly
After our discussion in #590 I was trying to adopt that changes to our codebase and I've noticed something unexpected. Still not sure is that intended or is it my misunderstanding, so I've decided to open a discussion
When my consumer .fetch() or .consume() message I got duplicates. In following screenshot there was 6 unprocessed messages and I've called .fetch() with default config (i.e. should fetch max 100 messages).

In my understanding I should receive 6 messages, but in reality I got 100 messages with lots of duplicates.
What I'm trying to implement:
Read N messages from nats, for every message call .working() to prevent re-deliveries, when I got N message or M time have passed and previous batch was processed - execute handlers for that messages in parallel.

[aricart]

Can you show me an example, and the options passed to create the consumer?

[whocaresk]

Sure, here:

import * as nats from 'nats';
import { JSONCodec } from 'nats';
import { randomUUID } from 'crypto';

const SERVERS = ['localhost:4222'];
const STREAM_NAME = 'stream';
const SUBJECT_NAME = 'str.>';

interface INatsMessage {
  uid: string;
  data: unknown;
}

(async () => {
  const nc = await nats.connect({
    user: 'dev',
    servers: SERVERS,
    pingInterval: 3000,
    reconnect: true,
    maxReconnectAttempts: -1,
    debug: true,
  });

  const iter = nc.status();
  (async () => {
    for await (const e of iter) {
      console.log('---EVENT: ', e);
    }
  })().then();

  const jsm = await nc.jetstreamManager();

  try {
    await jsm.streams.delete(STREAM_NAME);
  } catch (err) {}

  await jsm.streams.add({
    name: STREAM_NAME,
    subjects: [SUBJECT_NAME],
  });

  const decodedMessages: INatsMessage[] = [];
  const codec = JSONCodec<INatsMessage>();
  const subject = `str.messages.new`;
  const queue = `${subject}_NewMessageHandler`;
  const info = await jsm.consumers.add(STREAM_NAME, {
    durable_name: queue.replace(/\./g, '_'),
    ack_policy: nats.AckPolicy.Explicit,
    ack_wait: 1000 * 60 * 2,
    max_ack_pending: 20000,
  });

  const consumer = await jsm.jetstream().consumers.get(STREAM_NAME, info.name);

  for (let i = 0; i < 6; i++) {
    await jsm.jetstream().publish(
      subject,
      codec.encode({
        uid: randomUUID(),
        data: 'test',
      }),
      {
        msgID: randomUUID(),
      },
    );
  }

  const messages = await consumer.fetch();

  for await (const message of messages) {
    message.working();

    const decoded = codec.decode(message.data);

    decodedMessages.push(decoded);

    await message.ackAck();
  }

  console.log({ decodedMessages });
})();

[whocaresk]

@aricart Hi! Have you had a time to take a look at this?

[whocaresk]

@aricart Hi again :) I'm still waiting for you

[aricart]

@whocaresk sorry this fell out of my radar, didn't see the notification.
When I copied your code, I saw the same issue - and then digging, realized that the problem is that your ack wait option:
ack_wait: 1000 * 60 * 2 is effectively 120,000ns - the type for that setting is specified in nanos, so you need to wrap it with the provided nanos() function:
ack_wait: nanos(1000 * 60 * 2)

[whocaresk]

you're right, that was the cause. Thanks!
I've tried to change that value with no luck previously, but probably done something wrong

[aricart]

it was the units - when I copy/pasted your code, I didn't think about that. You have code hints enabled right? if you are using an IDE, the doc should be popping up. For node, the typescript is already compiled, so it tends to display as number instead of Nanos which is the typing it has - IDE like visual studio code, do a really good job:

Unfortunately, I didn't do a replacer on the configuration (it is straight up what the server processes) so I didn't do conversion between millis and nanos (which I could have done). At the time I thought that was an ambiguity, so I left it as it was, I needed to have hidden that property and exposed an ack_wait_millis and then converted that under the hood. I am going to think about it for the 3.x version of the libs.