nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: nats-io/nats-server
Pull request #1861 (2021-01-29 01:52:55)

js: consumer list request fix
Noticed in the logs a few of these entries when running tests:

```
Did not receive all stream info results
```

In case there are no consumers for a stream, sends response with the empty list without waiting.

Signed-off-by: Waldemar Quevedo <wally@synadia.com>

/cc @nats-io/core



Patch:
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index d1be8650..424facfb 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -2560,6 +2560,18 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 	defer s.sysUnsubscribe(rsub)
 
 	// Send out our requests here.
+	var resp = JSApiConsumerListResponse{
+		ApiResponse: ApiResponse{Type: JSApiConsumerListResponseType},
+		Consumers:   []*ConsumerInfo{},
+	}
+
+	if len(consumers) == 0 {
+		resp.Limit = JSApiListLimit
+		resp.Offset = offset
+		s.sendAPIResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(resp))
+		return
+	}
+
 	for _, ca := range consumers {
 		isubj := fmt.Sprintf(clusterConsumerInfoT, ca.Client.Account, stream, ca.Name)
 		s.sendInternalMsgLocked(isubj, inbox, nil, nil)
@@ -2569,11 +2581,6 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 	notActive := time.NewTimer(timeout)
 	defer notActive.Stop()
 
-	var resp = JSApiConsumerListResponse{
-		ApiResponse: ApiResponse{Type: JSApiConsumerListResponseType},
-		Consumers:   []*ConsumerInfo{},
-	}
-
 LOOP:
 	for {
 		select {
diff --git a/server/sublist.go b/server/sublist.go
index d47e91ef..8a019bd1 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -510,16 +510,10 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	start := 0
 	for i := 0; i < len(subject); i++ {
 		if subject[i] == btsep {
-			if i-start == 0 {
-				return emptyResult
-			}
 			tokens = append(tokens, subject[start:i])
 			start = i + 1
 		}
 	}
-	if start >= len(subject) {
-		return emptyResult
-	}
 	tokens = append(tokens, subject[start:])
 
 	// FIXME(dlc) - Make shared pool between sublist and client readLoop?


Test patch:
diff --git a/server/sublist_test.go b/server/sublist_test.go
index e17325cb..21cb27f3 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1396,32 +1396,6 @@ func TestSublistReverseMatch(t *testing.T) {
 	verifyMember(r.psubs, fooBarBazSub, t)
 }
 
-func TestSublistMatchWithEmptyTokens(t *testing.T) {
-	for _, test := range []struct {
-		name  string
-		cache bool
-	}{
-		{"cache", true},
-		{"no cache", false},
-	} {
-		t.Run(test.name, func(t *testing.T) {
-			sl := NewSublist(true)
-			sub1 := newSub(">")
-			sub2 := newQSub(">", "queue")
-			sl.Insert(sub1)
-			sl.Insert(sub2)
-
-			for _, subj := range []string{".foo", "..foo", "foo..", "foo.", "foo..bar", "foo...bar"} {
-				t.Run(subj, func(t *testing.T) {
-					r := sl.Match(subj)
-					verifyLen(r.psubs, 0, t)
-					verifyQLen(r.qsubs, 0, t)
-				})
-			}
-		})
-	}
-}
-
 // -- Benchmarks Setup --
 
 var benchSublistSubs []*subscription
Repository: nats-io/nats-server
Pull request #1944 (2021-02-27 01:53:40)

changing NEXT expires field from time to duration in nanoseconds
Signed-off-by: Matthias Hanel <mh@synadia.com>

Moved this into a separate branch as per @wallyqs suggestion.


Patch:
diff --git a/server/consumer.go b/server/consumer.go
index f1a335b1..c397c7f3 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1552,8 +1552,10 @@ func nextReqFromMsg(msg []byte) (time.Time, int, bool, error) {
 		if err := json.Unmarshal(req, &cr); err != nil {
 			return time.Time{}, -1, false, err
 		}
-		return cr.Expires, cr.Batch, cr.NoWait, nil
-
+		if cr.Expires == time.Duration(0) {
+			return time.Time{}, cr.Batch, cr.NoWait, nil
+		}
+		return time.Now().Add(cr.Expires), cr.Batch, cr.NoWait, nil
 	default:
 		if n, err := strconv.Atoi(string(req)); err == nil {
 			return time.Time{}, n, false, nil
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 6725384a..bc0ea0c8 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -505,9 +505,9 @@ const JSApiConsumerListResponseType = "io.nats.jetstream.api.v1.consumer_list_re
 
 // JSApiConsumerGetNextRequest is for getting next messages for pull based consumers.
 type JSApiConsumerGetNextRequest struct {
-	Expires time.Time `json:"expires,omitempty"`
-	Batch   int       `json:"batch,omitempty"`
-	NoWait  bool      `json:"no_wait,omitempty"`
+	Expires time.Duration `json:"expires,omitempty"`
+	Batch   int           `json:"batch,omitempty"`
+	NoWait  bool          `json:"no_wait,omitempty"`
 }
 
 // JSApiStreamTemplateCreateResponse for creating templates.


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index fefc38f1..8c95d294 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3596,6 +3596,16 @@ func TestJetStreamClusterSuperClusterCrossClusterConsumerInterest(t *testing.T)
 	checkSubsPending(t, sub, 2)
 }
 
+func TestJetStreamNextReqFromMsg(t *testing.T) {
+	bef := time.Now()
+	expires, _, _, err := nextReqFromMsg([]byte(`{"expires":5000000000}`)) // nanoseconds
+	require_NoError(t, err)
+	now := time.Now()
+	if expires.Before(bef.Add(5*time.Second)) || expires.After(now.Add(5*time.Second)) {
+		t.Fatal("Expires out of expected range")
+	}
+}
+
 func TestJetStreamClusterSuperClusterPeerReassign(t *testing.T) {
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 60016fbb..ddf410ed 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -1784,7 +1784,7 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			// Now do expiration
 			req.Batch = 1
 			req.NoWait = false
-			req.Expires = time.Now().Add(10 * time.Millisecond)
+			req.Expires = 10 * time.Millisecond
 			jreq, _ = json.Marshal(req)
 
 			nc.PublishRequest(getSubj, reply, jreq)