[FIXED] Consistent filtered consumer state reporting

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/consumer.go

@@ -2983,13 +2983,14 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		}
 	}
 
-	// If we are replicated, we need to pull certain data from our store.
-	if rg != nil && rg.node != nil && o.store != nil {
+	// We always need to pull certain data from our store.
+	if o.store != nil {
 		state, err := o.store.BorrowState()
 		if err != nil {
 			o.mu.Unlock()
 			return nil
 		}
+
 		// If we are the leader we could have o.sseq that is skipped ahead.
 		// To maintain consistency in reporting (e.g. jsz) we always take the state for our delivered/ackfloor stream sequence.
 		// Only use skipped ahead o.sseq if we're a new consumer and have not yet replicated this state yet.

server/jetstream_cluster_3_test.go

@@ -5474,14 +5474,9 @@ func TestJetStreamClusterConsumerMaxDeliveryNumAckPendingBug(t *testing.T) {
 		t.Helper()
 		require_Equal(t, a.Delivered.Consumer, 10)
 		require_Equal(t, a.Delivered.Stream, 10)
-		// If replicated, agreed upon state is used. Otherwise, o.asflr and o.adflr would be skipped ahead for R1.
-		if replicated {
-			require_Equal(t, a.AckFloor.Consumer, 0)
-			require_Equal(t, a.AckFloor.Stream, 0)
-		} else {
-			require_Equal(t, a.AckFloor.Consumer, 10)
-			require_Equal(t, a.AckFloor.Stream, 10)
-		}
+		// Agreed upon state is always used. Otherwise, o.asflr and o.adflr would be skipped ahead.
+		require_Equal(t, a.AckFloor.Consumer, 0)
+		require_Equal(t, a.AckFloor.Stream, 0)
 		require_Equal(t, a.NumPending, 40)
 		require_Equal(t, a.NumRedelivered, 10)
 		a.Cluster, b.Cluster = nil, nil

server/jetstream_consumer_test.go

@@ -6467,7 +6467,7 @@ func TestJetStreamConsumerPendingLowerThanStreamFirstSeq(t *testing.T) {
 	})
 	require_NoError(t, err)
 
-	for i := 0; i < 100; i++ {
+	for i := 0; i < 10; i++ {
 		sendStreamMsg(t, nc, "foo", "msg")
 	}
 
@@ -6482,7 +6482,8 @@ func TestJetStreamConsumerPendingLowerThanStreamFirstSeq(t *testing.T) {
 
 	sub := natsSubSync(t, nc, inbox)
 	for i := 0; i < 10; i++ {
-		natsNexMsg(t, sub, time.Second)
+		msg := natsNexMsg(t, sub, time.Second)
+		require_NoError(t, msg.AckSync())
 	}
 
 	acc, err := s.lookupAccount(globalAccountName)
@@ -6508,6 +6509,13 @@ func TestJetStreamConsumerPendingLowerThanStreamFirstSeq(t *testing.T) {
 	require_True(t, si.State.FirstSeq == 1_000_000)
 	require_True(t, si.State.LastSeq == 999_999)
 
+	acc, err = s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err = acc.lookupStream("TEST")
+	require_NoError(t, err)
+	o = mset.lookupConsumer("dur")
+	require_True(t, o != nil)
+
 	natsSubSync(t, nc, inbox)
 	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
 		ci, err := js.ConsumerInfo("TEST", "dur")
@@ -6517,8 +6525,18 @@ func TestJetStreamConsumerPendingLowerThanStreamFirstSeq(t *testing.T) {
 		if ci.NumAckPending != 0 {
 			return fmt.Errorf("NumAckPending should be 0, got %v", ci.NumAckPending)
 		}
-		if ci.Delivered.Stream != 999_999 {
-			return fmt.Errorf("Delivered.Stream should be 999,999, got %v", ci.Delivered.Stream)
+		// Delivered stays the same because it reflects what was actually delivered.
+		// And must not be influenced by purges/compacts.
+		if ci.Delivered.Stream != 10 {
+			return fmt.Errorf("Delivered.Stream should be 10, got %v", ci.Delivered.Stream)
+		}
+
+		// Starting sequence should be skipped ahead, respecting the compact.
+		o.mu.RLock()
+		sseq := o.sseq
+		o.mu.RUnlock()
+		if sseq != 1_000_000 {
+			return fmt.Errorf("o.sseq should be 1,000,000, got %v", sseq)
 		}
 		return nil
 	})
@@ -9531,3 +9549,60 @@ func TestJetStreamConsumerDeliverAllOverlappingFilterSubjects(t *testing.T) {
 		}
 	}
 }
+
+func TestJetStreamConsumerStateAlwaysFromStore(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.>"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:       "CONSUMER",
+		AckWait:       2 * time.Second,
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo.bar",
+	})
+	require_NoError(t, err)
+
+	// Publish two messages, one the consumer is interested in.
+	_, err = js.Publish("foo.bar", nil)
+	require_NoError(t, err)
+	_, err = js.Publish("foo.other", nil)
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("foo.bar", "CONSUMER")
+	require_NoError(t, err)
+	defer sub.Drain()
+
+	// Consumer info should start empty.
+	ci, err := js.ConsumerInfo("TEST", "CONSUMER")
+	require_NoError(t, err)
+	require_Equal(t, ci.Delivered.Stream, 0)
+	require_Equal(t, ci.AckFloor.Stream, 0)
+
+	// Fetch more messages than match our filter.
+	msgs, err := sub.Fetch(2, nats.MaxWait(time.Second))
+	require_NoError(t, err)
+	require_Len(t, len(msgs), 1)
+
+	// We have received, but not acknowledged, consumer info must reflect that.
+	ci, err = js.ConsumerInfo("TEST", "CONSUMER")
+	require_NoError(t, err)
+	require_Equal(t, ci.Delivered.Stream, 1)
+	require_Equal(t, ci.AckFloor.Stream, 0)
+
+	// Now we acknowledge the message and expect our delivered/ackfloor to be correct.
+	require_NoError(t, msgs[0].AckSync())
+
+	ci, err = js.ConsumerInfo("TEST", "CONSUMER")
+	require_NoError(t, err)
+	require_Equal(t, ci.Delivered.Stream, 1)
+	require_Equal(t, ci.AckFloor.Stream, 1)
+}