Jetstream: Message.Metadata().NumDelivered counter sometimes provide invalid value

[bazh]

I'm playing with Ack/Nak functionality for streams with interest policy enabled.
My goal is to stop and remove messages which are failed to process for N times:

...
const maxDeliver = 5

msgs, _ := cons.Fetch(1)

for msg := range msgs.Messages() {
	meta, _ := msg.Metadata()

	if meta.NumDelivered >= maxDeliver {
		msg.Term()
	} else {
		msg.NakWithDelay(nakDelay)
	}
}

I've noticed, that meta.NumDelivered sometimes provides wrong number for deliveries.
For example, if maxDeliver == 5, then meta.NumDelivered gives numbers 1, 2, 3, 1, 1 (I'm expecting 1, 2, 3, 4, 5)

I see such behavior when delay between redelivery attempts (via msg.NakWithDelay(nakDelay)) is 1 minute or more.
For 5 second delay, it seems to work correct

Please help me understand if I'm missing something?

Full snippet:

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

const streamName = "test_stream28"
const subj = "test28"
const maxDeliver = 5

var nakDelay = 1 * time.Minute

func main() {
	nc, _ := nats.Connect(nats.DefaultURL)
	js, _ := jetstream.New(nc)

	ctx := context.TODO()

	stream, _ := js.CreateStream(ctx, jetstream.StreamConfig{
		Subjects:  []string{subj},
		Name:      streamName,
		Retention: jetstream.InterestPolicy,
	})

	cons, _ := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable:       "test_consumer",
		AckPolicy:     jetstream.AckExplicitPolicy,
		DeliverPolicy: jetstream.DeliverNewPolicy,
		MaxDeliver:    maxDeliver,
	})

	go func() {
		for {
			msgs, _ := cons.Fetch(1)

			for msg := range msgs.Messages() {
				meta, _ := msg.Metadata()

				if meta.NumDelivered >= maxDeliver {
					fmt.Printf("term: %2d, seq: %2d %v\n", meta.NumDelivered, meta.Sequence.Consumer, msg.Data())
					msg.Term()
				} else {
					fmt.Printf(" nak: %2d, seq: %2d %v\n", meta.NumDelivered, meta.Sequence.Consumer, msg.Data())
					msg.NakWithDelay(nakDelay)
				}
			}
		}
	}()

	for i := 0; i < 25; i++ {
		nc.Publish(subj, []byte{byte(i + 1)})
		time.Sleep(nakDelay*maxDeliver + 2*time.Second)
		fmt.Println("--------------------------------")
	}

	ch := make(chan int, 1)
	<-ch
}

And the output is:

 nak:  1, seq:  1 [1]
 nak:  2, seq:  2 [1]
 nak:  1, seq:  3 [1]
 nak:  4, seq:  4 [1]
term:  5, seq:  5 [1]
--------------------------------
 nak:  1, seq:  6 [2]
 nak:  2, seq:  7 [2]
 nak:  3, seq:  8 [2]
 nak:  4, seq:  9 [2]
 nak:  1, seq: 10 [2]
--------------------------------
 nak:  1, seq: 11 [3]
 nak:  1, seq: 12 [3]
 nak:  3, seq: 13 [3]
 nak:  4, seq: 14 [3]
 nak:  1, seq: 15 [3]
--------------------------------
 nak:  1, seq: 16 [4]
^Csignal: interrupt

[Jarema]

When a message is not acked, it does not block the consumer until the message is redelivered.
Instead, other messages for a given consumer are delivered, and the non-acked one will be, when its ack wait duration will expire.

You can check that by checking the message payload if its unique (keep in mind that what you print as payload are bytes here, not string).

That is probably what you see here.

[ripienaar]

Log the stream sequence, that'll also make it clear - easier than logging payload.

[bazh]

In the snippet, messages are fired one by one:
Fire msg1, sleep for nakDelay*maxDeliver, then fire msg2 and so on...

I've extended log output with the stream sequence
What I see: message is redelivered MaxDeliver times (As configured in consumer), however sometimes
msg.Meta().NumDelivered gives wrong number like NumDelivered == 1 while it should be 2:

 nak: num_delivered:  1, seq.stream:  1, seq.consumer:  1 [1]
 nak: num_delivered:  2, seq.stream:  1, seq.consumer:  2 [1]
 nak: num_delivered:  3, seq.stream:  1, seq.consumer:  3 [1] <--- For this message num_delivered is correct
 nak: num_delivered:  4, seq.stream:  1, seq.consumer:  4 [1]
term: num_delivered:  5, seq.stream:  1, seq.consumer:  5 [1]
--------------------------------
 nak: num_delivered:  1, seq.stream:  2, seq.consumer:  6 [2]
 nak: num_delivered:  1, seq.stream:  2, seq.consumer:  7 [2] <--- num_delivered should be 2?
 nak: num_delivered:  3, seq.stream:  2, seq.consumer:  8 [2]
 nak: num_delivered:  4, seq.stream:  2, seq.consumer:  9 [2]
term: num_delivered:  5, seq.stream:  2, seq.consumer: 10 [2]
--------------------------------
 nak: num_delivered:  1, seq.stream:  3, seq.consumer: 11 [3]
 nak: num_delivered:  2, seq.stream:  3, seq.consumer: 12 [3]
 nak: num_delivered:  3, seq.stream:  3, seq.consumer: 13 [3]
 nak: num_delivered:  4, seq.stream:  3, seq.consumer: 14 [3]
term: num_delivered:  5, seq.stream:  3, seq.consumer: 15 [3]

Subscribing to advisory also shows deliveries as 1 (I suppose it should be 2?):

[#7] Received on "$JS.EVENT.ADVISORY.CONSUMER.MSG_NAKED.test_stream100.test_consumer"
{"type":"io.nats.jetstream.advisory.v1.nak","id":"gdZzjUF4rkZYlJsUR4clQz","timestamp":"2024-03-05T16:05:58.726539096Z","stream":"test_stream100","consumer":"test_consumer","consumer_seq":6,"stream_seq":2,"deliveries":1}


[#8] Received on "$JS.EVENT.ADVISORY.CONSUMER.MSG_NAKED.test_stream100.test_consumer"
{"type":"io.nats.jetstream.advisory.v1.nak","id":"gdZzjUF4rkZYlJsUR4clV1","timestamp":"2024-03-05T16:06:58.737049305Z","stream":"test_stream100","consumer":"test_consumer","consumer_seq":7,"stream_seq":2,"deliveries":1} <--- Here it is


[#9] Received on "$JS.EVENT.ADVISORY.CONSUMER.MSG_NAKED.test_stream100.test_consumer"
{"type":"io.nats.jetstream.advisory.v1.nak","id":"gdZzjUF4rkZYlJsUR4clZ3","timestamp":"2024-03-05T16:07:58.743395263Z","stream":"test_stream100","consumer":"test_consumer","consumer_seq":8,"stream_seq":2,"deliveries":3}


[#10] Received on "$JS.EVENT.ADVISORY.CONSUMER.MSG_NAKED.test_stream100.test_consumer"
{"type":"io.nats.jetstream.advisory.v1.nak","id":"gdZzjUF4rkZYlJsUR4cld5","timestamp":"2024-03-05T16:08:58.748899763Z","stream":"test_stream100","consumer":"test_consumer","consumer_seq":9,"stream_seq":2,"deliveries":4}


[#11] Received on "$JS.EVENT.ADVISORY.CONSUMER.MSG_TERMINATED.test_stream100.test_consumer"
{"type":"io.nats.jetstream.advisory.v1.terminated","id":"gdZzjUF4rkZYlJsUR4clh7","timestamp":"2024-03-05T16:09:58.756618096Z","stream":"test_stream100","consumer":"test_consumer","consumer_seq":10,"stream_seq":2,"deliveries":5}

The worst scenario:
We see invalid number for num_delivered in the last attempt of delivery, then NATS automatically terminate message processing as per NumDelivered setting, but we don't know about that

--------------------------------
 nak: num_delivered:  1, seq.stream:  4, seq.consumer: 16 [4]
 nak: num_delivered:  2, seq.stream:  4, seq.consumer: 17 [4]
 nak: num_delivered:  3, seq.stream:  4, seq.consumer: 18 [4]
 nak: num_delivered:  1, seq.stream:  4, seq.consumer: 19 [4]
 nak: num_delivered:  1, seq.stream:  4, seq.consumer: 20 [4] <--- Here it is

[Jarema]

Thanks for detailed response.

This does look suspicious, however after running some tests (with code very similar to yours) I cannot replicate it.

Can you share details about server version and setup (clustered, single server)?

[bazh]

I'm running a single instance of NATS via docker desktop on arm chipset:

docker run -it -p 4222:4222 --rm nats:latest -js
Version:  2.10.11
Git:      [5fc6051]

Also tried nightly build:

run -it -p 4222:4222 --rm synadia/nats-server:nightly  -js
Version:  2.11.0-dev
Git:      [a503d43]

Interestingly, that on my environment I see such strange behavior only when NakDelay is one minute or more
For 5 seconds delay it seems to work correct

[Jarema]

Ok. Thanks, I will investigate further. I'm testing with 2minuts NAK.