[FIXED] Make sure jsz reporting of filtered consumers delivered strea…

…m sequence is consistent. (#5528)

For a leader, it will skip msgs at the end to eof and remember o.sseq
for the next getNextMsg call.
But this will report stream delivered different for the leader vs the
followers.

Also if a consumer leader processed an ack ahead of delivered it would
sync, but followers would not.
This makes the behavior consistent between leaders and followers.

Signed-off-by: Derek Collison <[email protected]>

---------

Signed-off-by: Derek Collison <[email protected]>

server/consumer.go

@@ -2719,17 +2719,24 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		}
 	}
 
-	// If we are replicated and we are not the leader we need to pull certain data from our store.
-	if rg != nil && rg.node != nil && !o.isLeader() && o.store != nil {
+	// If we are replicated and we are not the leader or we are filtered, we need to pull certain data from our store.
+	isLeader := o.isLeader()
+	if rg != nil && rg.node != nil && o.store != nil && (!isLeader || o.isFiltered()) {
 		state, err := o.store.BorrowState()
 		if err != nil {
 			o.mu.Unlock()
 			return nil
 		}
-		info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream
-		info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream
-		info.NumAckPending = len(state.Pending)
-		info.NumRedelivered = len(state.Redelivered)
+		if !isLeader {
+			info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream
+			info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream
+			info.NumAckPending = len(state.Pending)
+			info.NumRedelivered = len(state.Redelivered)
+		} else {
+			// Since we are filtered and we are the leader we could have o.sseq that is skipped ahead.
+			// To maintain consistency in reporting (e.g. jsz) we take the state for our delivered stream sequence.
+			info.Delivered.Stream = state.Delivered.Stream
+		}
 	}
 
 	// Adjust active based on non-zero etc. Also make UTC here.

server/filestore.go

@@ -8758,6 +8758,12 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 		return nil
 	}
 
+	// Match leader logic on checking if ack is ahead of delivered.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq > o.state.Delivered.Stream {
+		o.state.Delivered.Stream = sseq + 1
+	}
+
 	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
 		return ErrStoreMsgNotFound
 	}

server/jetstream_cluster_1_test.go

@@ -5966,7 +5966,54 @@ func TestJetStreamClusterFailMirrorsAndSources(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterConsumerDeliveredSyncReporting(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.WorkQueuePolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("foo.bar", "mc")
+	require_NoError(t, err)
+
+	// Make us match first, but not next 10.
+	_, err = js.Publish("foo.bar", nil)
+	require_NoError(t, err)
+	for i := 0; i < 10; i++ {
+		_, err = js.Publish("foo.baz", nil)
+		require_NoError(t, err)
+	}
+
+	msgs, err := sub.Fetch(1)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 1)
+
+	// Now we want to make sure that jsz reporting will show the same
+	// state, including delivered, which will have skipped to the end.
+	// The skip can happen on several factors, but for here we just send
+	// another pull request which we will let fail.
+	_, err = sub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+	require_Error(t, err)
+
+	opts := &JSzOptions{Accounts: true, Streams: true, Consumer: true}
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(opts)
+		require_NoError(t, err)
+		ci := jsz.AccountDetails[0].Streams[0].Consumer[0]
+		require_Equal(t, ci.Delivered.Consumer, 1)
+		require_Equal(t, ci.Delivered.Stream, 1)
+	}
+}
+
 //
-// DO NOT ADD NEW TESTS IN THIS FILE
+// DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.
 //

server/memstore.go

@@ -1658,6 +1658,12 @@ func (o *consumerMemStore) UpdateAcks(dseq, sseq uint64) error {
 		return nil
 	}
 
+	// Match leader logic on checking if ack is ahead of delivered.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq > o.state.Delivered.Stream {
+		o.state.Delivered.Stream = sseq + 1
+	}
+
 	// Check for AckAll here.
 	if o.cfg.AckPolicy == AckAll {
 		sgap := sseq - o.state.AckFloor.Stream