Repository: nats-io/nats-server
Pull request #6110 (2024-11-12 07:02:29)

Improve pinning/unpinning in Priority Groups
After round of feedback, few changes were requested:

**Fix unpin advisory name**

Copy-paste mistake was made in unpin advisory name.

**Fix PinnedTS**

It was meant to return time when pinned happened, not the daedline.

**Add check for group name**

As the ADR for Priority Groups was [also updated](https://github.com/nats-io/nats-architecture-and-design/pull/311) after feedbach, this PR implements that.

cc @ripienaar @jnmoyne 

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>


Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 9080f48a..8a046819 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"math/rand"
 	"reflect"
+	"regexp"
 	"slices"
 	"strconv"
 	"strings"
@@ -41,6 +42,10 @@ const (
 	JSPullRequestNatsPinId    = "Nats-Pin-Id"
 )
 
+var (
+	validGroupName = regexp.MustCompile(`^[a-zA-Z0-9/_=-]{1,16}$`)
+)
+
 // Headers sent when batch size was completed, but there were remaining bytes.
 const JsPullRequestRemainingBytesT = "NATS/1.0 409 Batch Completed\r\n%s: %d\r\n%s: %d\r\n\r\n"
 
@@ -819,6 +824,9 @@ func checkConsumerCfg(
 			if group == _EMPTY_ {
 				return NewJSConsumerEmptyGroupNameError()
 			}
+			if !validGroupName.MatchString(group) {
+				return NewJSConsumerInvalidGroupNameError()
+			}
 		}
 	}
 	return nil
@@ -1625,9 +1633,9 @@ func (o *consumer) sendPinnedAdvisoryLocked(group string) {
 
 }
 func (o *consumer) sendUnpinnedAdvisoryLocked(group string, reason string) {
-	e := JSConsumerGroupUnPinnedAdvisory{
+	e := JSConsumerGroupUnpinnedAdvisory{
 		TypedEvent: TypedEvent{
-			Type: JSConsumerGroupPinnedAdvisoryType,
+			Type: JSConsumerGroupUnpinnedAdvisoryType,
 			ID:   nuid.Next(),
 			Time: time.Now().UTC(),
 		},
@@ -3062,18 +3070,30 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample b
 		return false
 	}
 
-	// Check if this ack is above the current pointer to our next to deliver.
-	// This could happen on a cooperative takeover with high speed deliveries.
-	if sseq >= o.sseq {
-		o.sseq = sseq + 1
-	}
-
 	mset := o.mset
 	if mset == nil || mset.closed.Load() {
 		o.mu.Unlock()
 		return false
 	}
 
+	// Check if this ack is above the current pointer to our next to deliver.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq >= o.sseq {
+		// Let's make sure this is valid.
+		// This is only received on the consumer leader, so should never be higher
+		// than the last stream sequence.
+		var ss StreamState
+		mset.store.FastState(&ss)
+		if sseq > ss.LastSeq {
+			o.srv.Warnf("JetStream consumer '%s > %s > %s' ACK sequence %d past last stream sequence of %d",
+				o.acc.Name, o.stream, o.name, sseq, ss.LastSeq)
+			// FIXME(dlc) - For 2.11 onwards should we return an error here to the caller?
+			o.mu.Unlock()
+			return false
+		}
+		o.sseq = sseq + 1
+	}
+
 	// Let the owning stream know if we are interest or workqueue retention based.
 	// If this consumer is clustered (o.node != nil) this will be handled by
 	// processReplicatedAck after the ack has propagated.
@@ -3529,7 +3549,7 @@ func (o *consumer) setPinnedTimer(priorityGroup string) {
 	} else {
 		o.pinnedTtl = time.AfterFunc(o.cfg.PinnedTTL, func() {
 			o.mu.Lock()
-			o.pinnedTS = time.Now().Add(o.cfg.PinnedTTL)
+			o.pinnedTS = time.Now()
 			o.currentPinId = _EMPTY_
 			o.sendUnpinnedAdvisoryLocked(priorityGroup, "timeout")
 			o.mu.Unlock()
diff --git a/server/errors.json b/server/errors.json
index 0111df42..29d9ee57 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1598,5 +1598,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerInvalidGroupNameErr",
+    "code": 400,
+    "error_code": 10162,
+    "description": "Valid name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
diff --git a/server/filestore.go b/server/filestore.go
index 0a04a7c4..3d14b918 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -3332,11 +3332,11 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
 
 			var t uint64
 			var havePartial bool
-			mb.fss.Iter(func(bsubj []byte, ss *SimpleState) bool {
+			IntersectStree[SimpleState](mb.fss, sl, func(bsubj []byte, ss *SimpleState) {
 				subj := bytesToString(bsubj)
-				if havePartial || !sl.HasInterest(subj) {
+				if havePartial {
 					// If we already found a partial then don't do anything else.
-					return !havePartial
+					return
 				}
 				if ss.firstNeedsUpdate {
 					mb.recalculateFirstForSubj(subj, ss.First, ss)
@@ -3347,7 +3347,6 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
 					// We matched but its a partial.
 					havePartial = true
 				}
-				return !havePartial
 			})
 
 			// See if we need to scan msgs here.
@@ -3432,11 +3431,8 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
 				}
 				// Mark fss activity.
 				mb.lsts = time.Now().UnixNano()
-				mb.fss.Iter(func(bsubj []byte, ss *SimpleState) bool {
-					if sl.HasInterest(bytesToString(bsubj)) {
-						adjust += ss.Msgs
-					}
-					return true
+				IntersectStree(mb.fss, sl, func(bsubj []byte, ss *SimpleState) {
+					adjust += ss.Msgs
 				})
 			}
 		} else {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ce3bee67..f73d2f9c 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3424,6 +3424,11 @@ func (s *Server) jsConsumerUnpinRequest(sub *subscription, c *client, _ *Account
 		return
 	}
 
+	if !validGroupName.MatchString(req.Group) {
+		resp.Error = NewJSConsumerInvalidGroupNameError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if s.JetStreamIsClustered() {
 		// Check to make sure the stream is assigned.
 		js, cc := s.getJetStreamCluster()
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index cf5321df..acf42fc6 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5089,6 +5089,7 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 					}
 					panic(err.Error())
 				}
+
 				if err = o.store.Update(state); err != nil {
 					o.mu.RLock()
 					s, acc, mset, name := o.srv, o.acc, o.mset, o.name
@@ -5101,17 +5102,10 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 				if mset := o.getStream(); mset != nil {
 					var ss StreamState
 					mset.store.FastState(&ss)
-					if err := o.checkStateForInterestStream(&ss); err == errAckFloorHigherThanLastSeq {
-						// Register pre-acks unless no state at all for the stream and we would create alot of pre-acks.
-						mset.mu.Lock()
-						// Only register if we have a valid FirstSeq.
-						if ss.FirstSeq > 0 {
-							for seq := ss.FirstSeq; seq < state.AckFloor.Stream; seq++ {
-								mset.registerPreAck(o, seq)
-							}
-						}
-						mset.mu.Unlock()
-					}
+					// We used to register preacks here if our ack floor was higher than the last sequence.
+					// Now when streams catch up they properly call checkInterestState() and periodically run this as well.
+					// If our states drift this could have allocated lots of pre-acks.
+					o.checkStateForInterestStream(&ss)
 				}
 			}
 
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 1d633955..5d5ab454 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -122,6 +122,9 @@ const (
 	// JSConsumerInvalidDeliverSubject invalid push consumer deliver subject
 	JSConsumerInvalidDeliverSubject ErrorIdentifier = 10112
 
+	// JSConsumerInvalidGroupNameErr Valid name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters
+	JSConsumerInvalidGroupNameErr ErrorIdentifier = 10162
+
 	// JSConsumerInvalidPolicyErrF Generic delivery policy error ({err})
 	JSConsumerInvalidPolicyErrF ErrorIdentifier = 10094
 
@@ -527,6 +530,7 @@ var (
 		JSConsumerHBRequiresPushErr:                {Code: 400, ErrCode: 10088, Description: "consumer idle heartbeat requires a push based consumer"},
 		JSConsumerInactiveThresholdExcess:          {Code: 400, ErrCode: 10153, Description: "consumer inactive threshold exceeds system limit of {limit}"},
 		JSConsumerInvalidDeliverSubject:            {Code: 400, ErrCode: 10112, Description: "invalid push consumer deliver subject"},
+		JSConsumerInvalidGroupNameErr:              {Code: 400, ErrCode: 10162, Description: "Valid name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters"},
 		JSConsumerInvalidPolicyErrF:                {Code: 400, ErrCode: 10094, Description: "{err}"},
 		JSConsumerInvalidPriorityGroupErr:          {Code: 400, ErrCode: 10160, Description: "Provided priority group does not exist for this consumer"},
 		JSConsumerInvalidSamplingErrF:              {Code: 400, ErrCode: 10095, Description: "failed to parse consumer sampling configuration: {err}"},
@@ -1087,6 +1091,16 @@ func NewJSConsumerInvalidDeliverSubjectError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSConsumerInvalidDeliverSubject]
 }
 
+// NewJSConsumerInvalidGroupNameError creates a new JSConsumerInvalidGroupNameErr error: "Valid name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters"
+func NewJSConsumerInvalidGroupNameError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerInvalidGroupNameErr]
+}
+
 // NewJSConsumerInvalidPolicyError creates a new JSConsumerInvalidPolicyErrF error: "{err}"
 func NewJSConsumerInvalidPolicyError(err error, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 3480337b..825fd5c3 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -284,10 +284,10 @@ type JSConsumerGroupPinnedAdvisory struct {
 	PinnedClientId string `json:"pinned_id"`
 }
 
-const JSStreamGroupUnPinnedAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_group_unpinned"
+const JSConsumerGroupUnpinnedAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_group_unpinned"
 
-// JSConsumerGroupUnPinnedAdvisory indicates that a pin was lost
-type JSConsumerGroupUnPinnedAdvisory struct {
+// JSConsumerGroupUnpinnedAdvisory indicates that a pin was lost
+type JSConsumerGroupUnpinnedAdvisory struct {
 	TypedEvent
 	Account  string `json:"account,omitempty"`
 	Stream   string `json:"stream"`
diff --git a/server/memstore.go b/server/memstore.go
index 55e98882..ee953e1f 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -719,10 +719,7 @@ func (ms *memStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject boo
 	var havePartial bool
 	var totalSkipped uint64
 	// We will track start and end sequences as we go.
-	ms.fss.Iter(func(subj []byte, fss *SimpleState) bool {
-		if !sl.HasInterest(bytesToString(subj)) {
-			return true
-		}
+	IntersectStree[SimpleState](ms.fss, sl, func(subj []byte, fss *SimpleState) {
 		if fss.firstNeedsUpdate {
 			ms.recalculateFirstForSubj(bytesToString(subj), fss.First, fss)
 		}
@@ -736,7 +733,6 @@ func (ms *memStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject boo
 		} else {
 			totalSkipped += fss.Msgs
 		}
-		return true
 	})
 
 	// If we did not encounter any partials we can return here.
diff --git a/server/store.go b/server/store.go
index 6d894b28..ed972922 100644
--- a/server/store.go
+++ b/server/store.go
@@ -295,12 +295,16 @@ type DeleteRange struct {
 }
 
 func (dr *DeleteRange) State() (first, last, num uint64) {
-	return dr.First, dr.First + dr.Num, dr.Num
+	deletesAfterFirst := dr.Num
+	if deletesAfterFirst > 0 {
+		deletesAfterFirst--
+	}
+	return dr.First, dr.First + deletesAfterFirst, dr.Num
 }
 
 // Range will range over all the deleted sequences represented by this block.
 func (dr *DeleteRange) Range(f func(uint64) bool) {
-	for seq := dr.First; seq <= dr.First+dr.Num; seq++ {
+	for seq := dr.First; seq < dr.First+dr.Num; seq++ {
 		if !f(seq) {
 			return
 		}
diff --git a/server/stream.go b/server/stream.go
index 729ba4c8..a7217a98 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1692,8 +1692,8 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account, pedantic boo
 
 // Config returns the stream's configuration.
 func (mset *stream) config() StreamConfig {
-	mset.mu.RLock()
-	defer mset.mu.RUnlock()
+	mset.cfgMu.RLock()
+	defer mset.cfgMu.RUnlock()
 	return mset.cfg
 }
 
@@ -3654,7 +3654,6 @@ func (mset *stream) resetSourceInfo() {
 	}
 }
 
-// Lock should be held.
 // This will do a reverse scan on startup or leader election
 // searching for the starting sequence number.
 // This can be slow in degenerative cases.
diff --git a/server/sublist.go b/server/sublist.go
index 657666f1..d34bcde1 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -20,6 +20,8 @@ import (
 	"sync"
 	"sync/atomic"
 	"unicode/utf8"
+
+	"github.com/nats-io/nats-server/v2/server/stree"
 )
 
 // Sublist is a routing mechanism to handle subject distribution and
@@ -1735,3 +1737,44 @@ func getAllNodes(l *level, results *SublistResult) {
 		getAllNodes(n.next, results)
 	}
 }
+
+// IntersectStree will match all items in the given subject tree that
+// have interest expressed in the given sublist. The callback will only be called
+// once for each subject, regardless of overlapping subscriptions in the sublist.
+func IntersectStree[T any](st *stree.SubjectTree[T], sl *Sublist, cb func(subj []byte, entry *T)) {
+	var _subj [255]byte
+	intersectStree(st, sl.root, _subj[:0], cb)
+}
+
+func intersectStree[T any](st *stree.SubjectTree[T], r *level, subj []byte, cb func(subj []byte, entry *T)) {
+	if r.numNodes() == 0 {
+		st.Match(subj, cb)
+		return
+	}
+	nsubj := subj
+	if len(nsubj) > 0 {
+		nsubj = append(subj, '.')
+	}
+	switch {
+	case r.fwc != nil:
+		// We've reached a full wildcard, do a FWC match on the stree at this point
+		// and don't keep iterating downward.
+		nsubj := append(nsubj, '>')
+		st.Match(nsubj, cb)
+	case r.pwc != nil:
+		// We've found a partial wildcard. We'll keep iterating downwards, but first
+		// check whether there's interest at this level (without triggering dupes) and
+		// match if so.
+		nsubj := append(nsubj, '*')
+		if len(r.pwc.psubs)+len(r.pwc.qsubs) > 0 && r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
+			st.Match(nsubj, cb)
+		}
+		intersectStree(st, r.pwc.next, nsubj, cb)
+	case r.numNodes() > 0:
+		// Normal node with subject literals, keep iterating.
+		for t, n := range r.nodes {
+			nsubj := append(nsubj, t...)
+			intersectStree(st, n.next, nsubj, cb)
+		}
+	}
+}


Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 14987a7a..dc413959 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -6664,6 +6664,63 @@ func TestJetStreamClusterMetaRecoveryConsumerCreateAndRemove(t *testing.T) {
 	}
 }
 
+// Make sure if we received acks that are out of bounds, meaning past our
+// last sequence or before our first that they are ignored and errored if applicable.
+func TestJetStreamConsumerAckOutOfBounds(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",
+		Subjects:  []string{"foo.*"},
+		Retention: nats.WorkQueuePolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 10; i++ {
+		_, err = js.Publish("foo.bar", []byte("OK"))
+		require_NoError(t, err)
+	}
+
+	sub, err := js.PullSubscribe("foo.*", "C")
+	require_NoError(t, err)
+
+	msgs, err := sub.Fetch(1)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 1)
+	msgs[0].AckSync()
+
+	// Now ack way past the last sequence.
+	_, err = nc.Request("$JS.ACK.TEST.C.1.10000000000.0.0.0", nil, 250*time.Millisecond)
+	require_Error(t, err, nats.ErrTimeout)
+
+	// Make sure that now changes happened to our state.
+	ci, err := js.ConsumerInfo("TEST", "C")
+	require_NoError(t, err)
+	require_Equal(t, ci.Delivered.Consumer, 1)
+	require_Equal(t, ci.Delivered.Stream, 1)
+	require_Equal(t, ci.AckFloor.Consumer, 1)
+	require_Equal(t, ci.AckFloor.Stream, 1)
+
+	s := c.consumerLeader("$G", "TEST", "C")
+	s.Shutdown()
+	s.WaitForShutdown()
+	c.restartServer(s)
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "C")
+
+	// Confirm new leader has same state for delivered and ack floor.
+	ci, err = js.ConsumerInfo("TEST", "C")
+	require_NoError(t, err)
+	require_Equal(t, ci.Delivered.Consumer, 1)
+	require_Equal(t, ci.Delivered.Stream, 1)
+	require_Equal(t, ci.AckFloor.Consumer, 1)
+	require_Equal(t, ci.AckFloor.Stream, 1)
+}
+
 //
 // 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.
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 74d30807..e3b4a736 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -3790,10 +3790,21 @@ func TestJetStreamClusterPendingRequestsInJsz(t *testing.T) {
 		require_NoError(t, nc.PublishMsg(msg))
 	}
 
-	jsz, err := metaleader.Jsz(nil)
-	require_NoError(t, err)
-	require_True(t, jsz.Meta != nil)
-	require_NotEqual(t, jsz.Meta.Pending, 0)
+	// We could check before above published messages are received,
+	// so allow some retries for pending messages to build up.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		jsz, err := metaleader.Jsz(nil)
+		if err != nil {
+			return err
+		}
+		if jsz.Meta == nil {
+			return errors.New("jsz.Meta == nil")
+		}
+		if jsz.Meta.Pending == 0 {
+			return errors.New("jsz.Meta.Pending == 0, expected pending requests")
+		}
+		return nil
+	})
 
 	snc, _ := jsClientConnect(t, c.randomServer(), nats.UserInfo("admin", "s3cr3t!"))
 	defer snc.Close()
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 5f405ad5..a566908a 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -1807,7 +1807,6 @@ func TestJetStreamConsumerUnpinPickDifferentRequest(t *testing.T) {
 	require_NoError(t, err)
 	pinId := msg.Header.Get("Nats-Pin-Id")
 	require_NotEqual(t, pinId, "")
-	fmt.Printf("INITIAL PIN: %v\n", msg.Header.Get("Nats-Pin-Id"))
 
 	reqPinned := JSApiConsumerGetNextRequest{Batch: 5, Expires: 15 * time.Second, PriorityGroup: PriorityGroup{
 		Group: "A",
@@ -2015,6 +2014,7 @@ func TestJetStreamConsumerUnpin(t *testing.T) {
 		{"unpin on missing stream", nc, "NOT_EXIST", "C", "A", &ApiError{ErrCode: uint16(JSStreamNotFoundErr)}},
 		{"unpin on missing consumer", nc, "TEST", "NOT_EXIST", "A", &ApiError{ErrCode: uint16(JSConsumerNotFoundErr)}},
 		{"unpin missing group", nc, "TEST", "C", "", &ApiError{ErrCode: uint16(JSInvalidJSONErr)}},
+		{"unpin bad group name", nc, "TEST", "C", "group    name\r\n", &ApiError{ErrCode: uint16(JSConsumerInvalidGroupNameErr)}},
 		{"ok unpin", nc, "TEST", "C", "A", nil},
 	} {
 		t.Run(test.name, func(t *testing.T) {
@@ -2368,6 +2368,26 @@ func TestJetStreamConsumerMultipleFitersWithStartDate(t *testing.T) {
 
 }
 
+func TestPriorityGroupNameRegex(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		group string
+		valid bool
+	}{
+		{"valid-short", "A", true},
+		{"valid-with-accepted-special-chars", "group/consumer=A", true},
+		{"empty", "", false},
+		{"with-space", "A B", false},
+		{"with-tab", "A   B", false},
+		{"too-long-name", "group-name-that-is-too-long", false},
+		{"line-termination", "\r\n", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			require_Equal(t, test.valid, validGroupName.MatchString(test.group))
+		})
+	}
+}
+
 func sendRequest(t *testing.T, nc *nats.Conn, reply string, req JSApiConsumerGetNextRequest) *nats.Subscription {
 	reqb, _ := json.Marshal(req)
 	replies, err := nc.SubscribeSync(reply)
diff --git a/server/norace_test.go b/server/norace_test.go
index 05ef611d..3472d16f 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -6703,7 +6703,7 @@ func TestNoRaceJetStreamClusterGhostConsumers(t *testing.T) {
 			cc := sjs.cluster
 			sa := cc.streams[globalAccountName]["TEST"]
 			var consumers []string
-			for cName, _ := range sa.consumers {
+			for cName := range sa.consumers {
 				consumers = append(consumers, cName)
 			}
 			sjs.mu.Unlock()
diff --git a/server/store_test.go b/server/store_test.go
index fdf4690a..f7832974 100644
--- a/server/store_test.go
+++ b/server/store_test.go
@@ -109,3 +109,35 @@ func TestStoreMsgLoadNextMsgMulti(t *testing.T) {
 		},
 	)
 }
+
+func TestStoreDeleteSlice(t *testing.T) {
+	ds := DeleteSlice{2}
+	var deletes []uint64
+	ds.Range(func(seq uint64) bool {
+		deletes = append(deletes, seq)
+		return true
+	})
+	require_Len(t, len(deletes), 1)
+	require_Equal(t, deletes[0], 2)
+
+	first, last, num := ds.State()
+	require_Equal(t, first, 2)
+	require_Equal(t, last, 2)
+	require_Equal(t, num, 1)
+}
+
+func TestStoreDeleteRange(t *testing.T) {
+	dr := DeleteRange{First: 2, Num: 1}
+	var deletes []uint64
+	dr.Range(func(seq uint64) bool {
+		deletes = append(deletes, seq)
+		return true
+	})
+	require_Len(t, len(deletes), 1)
+	require_Equal(t, deletes[0], 2)
+
+	first, last, num := dr.State()
+	require_Equal(t, first, 2)
+	require_Equal(t, last, 2)
+	require_Equal(t, num, 1)
+}
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 2bc3a995..5e8145c5 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/nats-server/v2/server/stree"
 	"github.com/nats-io/nuid"
 )
 
@@ -1982,6 +1983,128 @@ func TestSublistNumInterest(t *testing.T) {
 	sl.Remove(qsub)
 }
 
+func TestSublistInterestBasedIntersection(t *testing.T) {
+	st := stree.NewSubjectTree[struct{}]()
+	st.Insert([]byte("one.two.three.four"), struct{}{})
+	st.Insert([]byte("one.two.three.five"), struct{}{})
+	st.Insert([]byte("one.two.six"), struct{}{})
+	st.Insert([]byte("one.two.seven"), struct{}{})
+	st.Insert([]byte("eight.nine"), struct{}{})
+
+	require_NoDuplicates := func(t *testing.T, got map[string]int) {
+		for _, c := range got {
+			require_Equal(t, c, 1)
+		}
+	}
+
+	t.Run("Literals", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one.two.six"))
+		sl.Insert(newSub("eight.nine"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 2)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("PWC", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one.two.*.*"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 2)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("PWCOverlapping", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one.two.*.four"))
+		sl.Insert(newSub("one.two.*.*"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 2)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("PWCAll", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("*.*"))
+		sl.Insert(newSub("*.*.*"))
+		sl.Insert(newSub("*.*.*.*"))
+		require_True(t, sl.HasInterest("foo.bar"))
+		require_True(t, sl.HasInterest("foo.bar.baz"))
+		require_True(t, sl.HasInterest("foo.bar.baz.qux"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 5)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("FWC", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one.>"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 4)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("FWCOverlapping", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one.two.three.four"))
+		sl.Insert(newSub("one.>"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 4)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("FWCAll", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub(">"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 5)
+		require_NoDuplicates(t, got)
+	})
+
+	t.Run("NoMatch", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 0)
+	})
+
+	t.Run("NoMatches", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("one"))
+		sl.Insert(newSub("eight"))
+		sl.Insert(newSub("ten"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 0)
+	})
+}
+
 func subsInit(pre string, toks []string) {
 	var sub string
 	for _, t := range toks {