Repository: nats-io/nats-server (go)

Issue:
Consumer using multiple "filter_subjects" is not receiving all messages when combined with "by_start_time" deliver policy
### Observed behavior

**Summary and example of the problem:**

I have `test` stream which contains exactly three messages in following subjects:
- `events.subjectOne`
- `events.subjectTwo`
- `events.subjectThree`

I have two consumers created on this stream:
- first consumer uses `filter_subjects: ["events.*"]` pattern,
- second one uses explicit `filter_subjects: ["events.subjectOne","events.subjectTwo","events.subjectThree",]`

After creating consumers first of them has `num_pending` messages set to 3, which is correct value, but second one receives `num_pending` as 2, even though it should also receive all messages. This only happens when consumers have set `deliver_policy` to `by_start_time`.

Similar behavior could be observed using different subject combinations, but probably at least 2 different subjects are required with multiple messages. When only two messages exists both consumers have `num_pending` set to 2, which is correct.

**Detailed messages received using `nats sub ">"`**

```
Stream creation:

[#7] Received on "$JS.API.STREAM.CREATE.test" with reply "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ45T"
{"name":"test","subjects":["events.>"],"max_msgs_per_subject":10000000,"max_msgs":10000000,"deny_purge":false,"discard":"old","max_age":604800000000000,"retention":"limits","num_replicas":1,"duplicate_window":300000000000}

[#9] Received on "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ45T"
{"type":"io.nats.jetstream.api.v1.stream_create_response","config":{"name":"test","subjects":["events.\u003e"],"retention":"limits","max_consumers":-1,"max_msgs":10000000,"max_bytes":-1,"max_age":604800000000000,"max_msgs_per_subject":10000000,"max_msg_size":-1,"discard":"old","storage":"file","num_replicas":1,"duplicate_window":300000000000,"compression":"none","allow_direct":false,"mirror_direct":false,"sealed":false,"deny_delete":false,"deny_purge":false,"allow_rollup_hdrs":false,"consumer_limits":{}},"created":"2024-11-05T09:42:37.025040829Z","state":{"messages":0,"bytes":0,"first_seq":0,"first_ts":"0001-01-01T00:00:00Z","last_seq":0,"last_ts":"0001-01-01T00:00:00Z","consumer_count":0},"ts":"2024-11-05T09:42:37.025677661Z","did_create":true}


Messages in stream:

[#11] Received on "events.subjectOne"
test1

[#12] Received on "events.subjectTwo"
test2

[#13] Received on "events.subjectThree"
test3


First consumer:

[#14] Received on "$JS.API.CONSUMER.DURABLE.CREATE.test.wildcard-consumer" with reply "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ49Y"
{"config":{"durable_name":"wildcard-consumer","filter_subjects":["events.*"],"deliver_policy":"by_start_time","opt_start_time":"2024-11-05T08:42:36.947Z"},"stream_name":"test","action":"create"}

[#16] Received on "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ49Y"
{"type":"io.nats.jetstream.api.v1.consumer_create_response","stream_name":"test","name":"wildcard-consumer","created":"2024-11-05T09:42:37.028304635Z","config":{"durable_name":"wildcard-consumer","deliver_policy":"by_start_time","opt_start_time":"2024-11-05T08:42:36.947Z","ack_policy":"none","max_deliver":-1,"filter_subjects":["events.*"],"replay_policy":"instant","max_waiting":512,"num_replicas":0},"delivered":{"consumer_seq":0,"stream_seq":0},"ack_floor":{"consumer_seq":0,"stream_seq":0},"num_ack_pending":0,"num_redelivered":0,"num_waiting":0,"num_pending":3,"ts":"2024-11-05T09:42:37.028627946Z"}


Second consumer:

[#18] Received on "$JS.API.CONSUMER.DURABLE.CREATE.test.explicit-subjects-consumer" with reply "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ4E3"
{"config":{"durable_name":"explicit-subjects-consumer","filter_subjects":["events.subjectOne","events.subjectTwo","events.subjectThree"],"deliver_policy":"by_start_time","opt_start_time":"2024-11-05T08:42:36.947Z"},"stream_name":"test","action":"create"}

[#20] Received on "_INBOX.1W01W6HG0ZCF0JTK7PJ3XJ.1W01W6HG0ZCF0JTK7PJ4E3"
{"type":"io.nats.jetstream.api.v1.consumer_create_response","stream_name":"test","name":"explicit-subjects-consumer","created":"2024-11-05T09:42:37.029950864Z","config":{"durable_name":"explicit-subjects-consumer","deliver_policy":"by_start_time","opt_start_time":"2024-11-05T08:42:36.947Z","ack_policy":"none","max_deliver":-1,"filter_subjects":["events.subjectOne","events.subjectTwo","events.subjectThree"],"replay_policy":"instant","max_waiting":512,"num_replicas":0},"delivered":{"consumer_seq":0,"stream_seq":1},"ack_floor":{"consumer_seq":0,"stream_seq":0},"num_ack_pending":0,"num_redelivered":0,"num_waiting":0,"num_pending":2,"ts":"2024-11-05T09:42:37.030212597Z"}
```


### Expected behavior

In example above second consumer should have `num_pending` set to 3, as its filter covers all existing messages. Due to wrong `num_pending` number second consumer is not receiving all messages.

### Server and client version

nats-server in Docker: nats:2.10.18-alpine3.20
nats-cli: 0.0.35
nats client Deno: nats@v1.28.2

### Host environment

Windows 10 Pro 22H2

### Steps to reproduce

See simple script written in Deno which is reproducing this behavior:
[nats-server-bug.zip](https://github.com/user-attachments/files/17630981/nats-server-bug.zip)


Pull request description:
Fix consumer starting sequence with start time and multiple filters
fixes #6076 

When starting by time option was used , a different codepath was taken when calculating starting sequence for the consumer. That codepath had off by one bug.

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>

Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 7b88c4e3..dcfb70ba 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -60,14 +60,8 @@ type ConsumerInfo struct {
 	Paused         bool            `json:"paused,omitempty"`
 	PauseRemaining time.Duration   `json:"pause_remaining,omitempty"`
 	// TimeStamp indicates when the info was gathered
-	TimeStamp      time.Time            `json:"ts"`
-	PriorityGroups []PriorityGroupState `json:"priority_groups,omitempty"`
-}
-
-type PriorityGroupState struct {
-	Group          string    `json:"group"`
-	PinnedClientID string    `json:"pinned_client_id,omitempty"`
-	PinnedTS       time.Time `json:"pinned_ts,omitempty"`
+	TimeStamp time.Time         `json:"ts"`
+	PinnedIds map[string]string `json:"pinned_ids,omitempty"`
 }
 
 type ConsumerConfig struct {
@@ -486,7 +480,6 @@ type consumer struct {
 	currentPinId string
 	/// pinnedTtl is the remaining time before the current PinId expires.
 	pinnedTtl *time.Timer
-	pinnedTS  time.Time
 }
 
 // A single subject filter.
@@ -2907,14 +2900,10 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		rg = o.ca.Group
 	}
 
-	priorityGroups := []PriorityGroupState{}
 	// TODO(jrm): when we introduce supporting many priority groups, we need to update assigning `o.currentNuid` for each group.
+	pinnedIds := make(map[string]string)
 	if len(o.cfg.PriorityGroups) > 0 {
-		priorityGroups = append(priorityGroups, PriorityGroupState{
-			Group:          o.cfg.PriorityGroups[0],
-			PinnedClientID: o.currentPinId,
-			PinnedTS:       o.pinnedTS,
-		})
+		pinnedIds[o.cfg.PriorityGroups[0]] = o.currentPinId
 	}
 
 	cfg := o.cfg
@@ -2936,7 +2925,7 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		NumPending:     o.checkNumPending(),
 		PushBound:      o.isPushMode() && o.active,
 		TimeStamp:      time.Now().UTC(),
-		PriorityGroups: priorityGroups,
+		PinnedIds:      pinnedIds,
 	}
 	if o.cfg.PauseUntil != nil {
 		p := *o.cfg.PauseUntil
@@ -3526,21 +3515,6 @@ func (o *consumer) pendingRequests() map[string]*waitingRequest {
 	return m
 }
 
-func (o *consumer) setPinnedTimer(priorityGroup string) {
-	if o.pinnedTtl != nil {
-		o.pinnedTtl.Reset(o.cfg.PinnedTTL)
-	} else {
-		o.pinnedTtl = time.AfterFunc(o.cfg.PinnedTTL, func() {
-			o.mu.Lock()
-			o.pinnedTS = time.Now().Add(o.cfg.PinnedTTL)
-			o.currentPinId = _EMPTY_
-			o.sendUnpinnedAdvisoryLocked(priorityGroup, "timeout")
-			o.mu.Unlock()
-			o.signalNewMessages()
-		})
-	}
-}
-
 // Return next waiting request. This will check for expirations but not noWait or interest.
 // That will be handled by processWaiting.
 // Lock should be held.
@@ -3591,8 +3565,6 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 				if wr.priorityGroup.Id == _EMPTY_ {
 					o.currentPinId = nuid.Next()
 					wr.priorityGroup.Id = o.currentPinId
-					o.setPinnedTimer(priorityGroup)
-
 				} else {
 					// There is pin id set, but not a matching one. Send a notification to the client and remove the request.
 					// Probably this is the old pin id.
@@ -3803,12 +3775,21 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 			return
 		}
 
-		if o.currentPinId != _EMPTY_ {
-			if priorityGroup.Id == o.currentPinId {
-				o.setPinnedTimer(priorityGroup.Group)
-			} else if priorityGroup.Id != _EMPTY_ {
-				sendErr(423, "Nats-Pin-Id mismatch")
-				return
+		if priorityGroup.Id != _EMPTY_ && priorityGroup.Id != o.currentPinId && o.currentPinId != _EMPTY_ {
+			sendErr(423, "Nats-Pin-Id mismatch")
+			return
+		} else {
+			if o.pinnedTtl != nil && priorityGroup.Id == o.currentPinId && o.currentPinId != _EMPTY_ {
+
+				o.pinnedTtl.Reset(o.cfg.PinnedTTL)
+			} else if o.pinnedTtl == nil {
+				o.pinnedTtl = time.AfterFunc(o.cfg.PinnedTTL, func() {
+					o.mu.Lock()
+					o.currentPinId = _EMPTY_
+					o.sendUnpinnedAdvisoryLocked(priorityGroup.Group, "timeout")
+					o.mu.Unlock()
+					o.signalNewMessages()
+				})
 			}
 		}
 	}
@@ -5350,7 +5331,7 @@ func (o *consumer) selectStartingSeqNo() {
 					for _, filter := range o.subjf {
 						// Use first sequence since this is more optimized atm.
 						ss := o.mset.store.FilteredState(state.FirstSeq, filter.subject)
-						if ss.First > o.sseq && ss.First < nseq {
+						if ss.First >= o.sseq && ss.First < nseq {
 							nseq = ss.First
 						}
 					}


Test patch:
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 0e97cff0..3e3bbdbb 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -1858,85 +1858,6 @@ func TestJetStreamConsumerUnpinPickDifferentRequest(t *testing.T) {
 	require_NotEqual(t, msg.Header.Get("Nats-Pin-Id"), "")
 }
 
-func TestJetStreamPinnedTTL(t *testing.T) {
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	nc, _ := jsClientConnect(t, s)
-	defer nc.Close()
-
-	acc := s.GlobalAccount()
-
-	mset, err := acc.addStream(&StreamConfig{
-		Name:      "TEST",
-		Subjects:  []string{"foo"},
-		Retention: LimitsPolicy,
-	})
-	require_NoError(t, err)
-
-	_, err = mset.addConsumer(&ConsumerConfig{
-		Durable:        "C",
-		FilterSubject:  "foo",
-		PriorityGroups: []string{"A"},
-		PriorityPolicy: PriorityPinnedClient,
-		AckPolicy:      AckExplicit,
-		PinnedTTL:      3 * time.Second,
-	})
-	require_NoError(t, err)
-
-	for i := 0; i < 10; i++ {
-		sendStreamMsg(t, nc, "foo", "data")
-	}
-
-	req := JSApiConsumerGetNextRequest{Batch: 1, Expires: 10 * time.Second, PriorityGroup: PriorityGroup{
-		Group: "A",
-	}}
-
-	reqBytes, err := json.Marshal(req)
-	require_NoError(t, err)
-
-	firstInbox := "FIRST"
-	firstReplies, err := nc.SubscribeSync(firstInbox)
-	require_NoError(t, err)
-	nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.C", firstInbox, reqBytes)
-
-	msg, err := firstReplies.NextMsg(1 * time.Second)
-	require_NoError(t, err)
-	pinId := msg.Header.Get("Nats-Pin-Id")
-	require_NotEqual(t, pinId, "")
-
-	secondInbox := "SECOND"
-	secondReplies, err := nc.SubscribeSync(secondInbox)
-	require_NoError(t, err)
-	nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.C", secondInbox, reqBytes)
-
-	// Expect error, as first request should be still pinned.
-	_, err = secondReplies.NextMsg(1 * time.Second)
-	require_Error(t, err)
-
-	// During the 5 second window, the first Pin should time out and this request
-	// should become the pinned one and get the message.
-	msg, err = secondReplies.NextMsg(5 * time.Second)
-	require_NoError(t, err)
-	newPinId := msg.Header.Get("Nats-Pin-Id")
-	require_NotEqual(t, newPinId, pinId)
-	require_NotEqual(t, newPinId, "")
-
-	thirdInbox := "THIRD"
-	thirdReplies, err := nc.SubscribeSync(thirdInbox)
-	require_NoError(t, err)
-	nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.C", thirdInbox, reqBytes)
-
-	// The same process as above, but tests different codepath - one where Pin
-	// is set on existing waiting request.
-	msg, err = thirdReplies.NextMsg(5 * time.Second)
-	require_NoError(t, err)
-	require_NotEqual(t, msg.Header.Get("Nats-Pin-Id"), pinId)
-	require_NotEqual(t, msg.Header.Get("Nats-Pin-Id"), newPinId)
-	require_NotEqual(t, newPinId, "")
-
-}
-
 func TestJetStreamConsumerUnpin(t *testing.T) {
 	single := RunBasicJetStreamServer(t)
 	defer single.Shutdown()
@@ -2315,6 +2236,56 @@ func TestJetStreamConsumerOverflow(t *testing.T) {
 	msg, err = maxAckPending50.NextMsg(time.Second)
 	require_NoError(t, err)
 	require_NotNil(t, msg)
+}
+
+func TestJetStreamConsumerMultipleFitersWithStartDate(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	past := time.Now().Add(-90 * time.Second)
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"events.>"},
+	})
+	require_NoError(t, err)
+
+	sendStreamMsg(t, nc, "events.foo", "msg-1")
+	sendStreamMsg(t, nc, "events.bar", "msg-2")
+	sendStreamMsg(t, nc, "events.baz", "msg-3")
+	sendStreamMsg(t, nc, "events.biz", "msg-4")
+	sendStreamMsg(t, nc, "events.faz", "msg-5")
+	sendStreamMsg(t, nc, "events.foo", "msg-6")
+	sendStreamMsg(t, nc, "events.biz", "msg-7")
+
+	for _, test := range []struct {
+		name                   string
+		filterSubjects         []string
+		startTime              time.Time
+		expectedMessages       uint64
+		expectedStreamSequence uint64
+	}{
+		{"Single-Filter-first-sequence", []string{"events.foo"}, past, 2, 0},
+		{"Multiple-Filter-first-sequence", []string{"events.foo", "events.bar", "events.baz"}, past, 4, 0},
+		{"Multiple-Filters-second-subject", []string{"events.bar", "events.baz"}, past, 2, 1},
+		{"Multiple-Filters-first-last-subject", []string{"events.foo", "events.biz"}, past, 4, 0},
+		{"Multiple-Filters-in-future", []string{"events.foo", "events.biz"}, time.Now().Add(1 * time.Minute), 0, 7},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			info, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+				Durable:        test.name,
+				FilterSubjects: test.filterSubjects,
+				DeliverPolicy:  nats.DeliverByStartTimePolicy,
+				OptStartTime:   &test.startTime,
+			})
+			require_NoError(t, err)
+			require_Equal(t, test.expectedStreamSequence, info.Delivered.Stream)
+			require_Equal(t, test.expectedMessages, info.NumPending)
+		})
+	}
 
 }