Cherry picks for v2.10.11 release

server/consumer.go

@@ -4783,6 +4783,23 @@ func (o *consumer) selectStartingSeqNo() {
 				// If we are here we are time based.
 				// TODO(dlc) - Once clustered can't rely on this.
 				o.sseq = o.mset.store.GetSeqFromTime(*o.cfg.OptStartTime)
+				// Here we want to see if we are filtered, and if so possibly close the gap
+				// to the nearest first given our starting sequence from time. This is so we do
+				// not force the system to do a linear walk between o.sseq and the real first.
+				if len(o.subjf) > 0 {
+					nseq := state.LastSeq
+					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 {
+							nseq = ss.First
+						}
+					}
+					// Skip ahead if possible.
+					if nseq > o.sseq && nseq < state.LastSeq {
+						o.sseq = nseq
+					}
+				}
 			} else {
 				// DeliverNew
 				o.sseq = state.LastSeq + 1

server/filestore.go

@@ -2246,22 +2246,33 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 	if !isAll && len(mb.fss) == 1 {
 		_, isAll = mb.fss[filter]
 	}
-	// Skip scan of mb.fss if number of messages in the block are less than
-	// 1/2 the number of subjects in mb.fss. Or we have a wc and lots of fss entries.
-	const linearScanMaxFSS = 32
 	// Make sure to start at mb.first.seq if fseq < mb.first.seq
 	if seq := atomic.LoadUint64(&mb.first.seq); seq > fseq {
 		fseq = seq
 	}
 	lseq := atomic.LoadUint64(&mb.last.seq)
-	doLinearScan := isAll || 2*int(lseq-fseq) < len(mb.fss) || (wc && len(mb.fss) > linearScanMaxFSS)
 
+	// Optionally build the isMatch for wildcard filters.
+	tsa := [32]string{}
+	fsa := [32]string{}
+	var fts []string
+	var isMatch func(subj string) bool
+	// Decide to build.
+	if wc {
+		fts = tokenizeSubjectIntoSlice(fsa[:0], filter)
+		isMatch = func(subj string) bool {
+			tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
+			return isSubsetMatchTokenized(tts, fts)
+		}
+	}
+	// Only do linear scan if isAll or we are wildcarded and have to traverse more fss than actual messages.
+	doLinearScan := isAll || (wc && len(mb.fss) > int(lseq-fseq))
 	if !doLinearScan {
 		// If we have a wildcard match against all tracked subjects we know about.
 		if wc {
 			subs = subs[:0]
 			for subj := range mb.fss {
-				if subjectIsSubsetMatch(subj, filter) {
+				if isMatch(subj) {
 					subs = append(subs, subj)
 				}
 			}
@@ -2283,6 +2294,13 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		}
 	}
 
+	// If we guess to not do a linear scan, but the above resulted in alot of subs that will
+	// need to be checked for every scanned message, revert.
+	// TODO(dlc) - we could memoize the subs across calls.
+	if len(subs) > int(lseq-fseq) {
+		doLinearScan = true
+	}
+
 	if fseq > lseq {
 		return nil, didLoad, ErrStoreMsgNotFound
 	}
@@ -2310,7 +2328,7 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 			return fsm, expireOk, nil
 		}
 		if doLinearScan {
-			if wc && subjectIsSubsetMatch(fsm.subj, filter) {
+			if wc && isMatch(sm.subj) {
 				return fsm, expireOk, nil
 			} else if !wc && fsm.subj == filter {
 				return fsm, expireOk, nil