Messages are skipped (disappeared) when using Rollup and multi-filter consumers

[yaarasat]

Observed behavior

Messages are getting disappeared. After long investigatin it was found out that this is happening when using rollups. To be more specific, for multiple filter subjects and having rollup, msgs are getting lost, It seems that nuts move the cursor to the last msg that was purged. and if there were msgs for other filter that were not purged, they are getting skipped since the cursor was moved after them.
At the end pending and redelivery counts become zero.

Expected behavior

When using rollup with multiple filters, only msgs that were supposed to be deleted are getting deleted. All other msgs should be consumed.

Server and client version

•  nats-server/v2 v2.14.3
•  nats.go v1.51.0

Host environment

No response

Steps to reproduce

1. Create a rollup-enabled stream for events.> .
   2. Create an explicit-ACK consumer with filters events.host and events.periodic , and max_ack_pending=1 .
   3. Publish events.host message A and hold its ACK.
   4. Publish events.host message B.
   5. Publish events.periodic with Nats-Rollup: sub .
   6. ACK message A.

Message B is skipped, the Periodic message is delivered, and the consumer reports no pending or redelivered messages.

Submission acknowledgment

• I am a human being writing in my own words and not an AI agent. I will not use an AI agent to communicate on my behalf in this issue, either directly or via copy-paste.

[yaarasat]

Test code to reproduce the issue:

package server

import (
	"context"
	"testing"
	"time"

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

func TestSubjectRollupDoesNotSkipOtherConsumerFilters(t *testing.T) {
	s := RunBasicJetStreamServer(t)
	defer s.Shutdown()

	nc, err := nats.Connect(s.ClientURL())
	require_NoError(t, err)
	defer nc.Close()

	js, err := jetstream.New(nc)
	require_NoError(t, err)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
		Name:        "EVENTS",
		Subjects:    []string{"events.>"},
		Retention:   jetstream.InterestPolicy,
		Storage:     jetstream.MemoryStorage,
		AllowRollup: true,
	})
	require_NoError(t, err)

	consumer, err := js.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{
		Name:           "consumer",
		Durable:        "consumer",
		AckPolicy:      jetstream.AckExplicitPolicy,
		FilterSubjects: []string{"events.host", "events.periodic"},
	})
	require_NoError(t, err)

	_, err = js.Publish(ctx, "events.host", []byte("host"))
	require_NoError(t, err)
	_, err = js.PublishMsg(ctx, &nats.Msg{
		Subject: "events.periodic",
		Header:  nats.Header{"Nats-Rollup": []string{"sub"}},
		Data:    []byte("periodic"),
	})
	require_NoError(t, err)

	batch, err := consumer.Fetch(2, jetstream.FetchMaxWait(time.Second))
	require_NoError(t, err)

	var subjects []string
	for msg := range batch.Messages() {
		subjects = append(subjects, msg.Subject())
		require_NoError(t, msg.Ack())
	}
	require_NoError(t, batch.Error())

	expected := []string{"events.host", "events.periodic"}
	if len(subjects) != len(expected) {
		t.Fatalf("expected subjects %v, got %v", expected, subjects)
	}
	for i := range expected {
		if subjects[i] != expected[i] {
			t.Fatalf("expected subjects %v, got %v", expected, subjects)
		}
	}
}