[IMPROVED] Improved time to select skip list and starting sequence number for deliver last by subject

server/consumer.go

@@ -4657,16 +4657,6 @@ type lastSeqSkipList struct {
 	seqs   []uint64
 }
 
-// Will create a skip list for us from a store's subjects state.
-func createLastSeqSkipList(mss map[string]SimpleState) []uint64 {
-	seqs := make([]uint64, 0, len(mss))
-	for _, ss := range mss {
-		seqs = append(seqs, ss.Last)
-	}
-	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
-	return seqs
-}
-
 // Let's us know we have a skip list, which is for deliver last per subject and we are just starting.
 // Lock should be held.
 func (o *consumer) hasSkipListPending() bool {
@@ -4697,38 +4687,36 @@ func (o *consumer) selectStartingSeqNo() {
 					}
 				}
 			} else if o.cfg.DeliverPolicy == DeliverLastPerSubject {
+				lss := &lastSeqSkipList{resume: state.LastSeq}
+				var smv StoreMsg
+				var filters []string
 				if o.subjf == nil {
-					if mss := o.mset.store.SubjectsState(o.cfg.FilterSubject); len(mss) > 0 {
-						o.lss = &lastSeqSkipList{
-							resume: state.LastSeq,
-							seqs:   createLastSeqSkipList(mss),
-						}
-						o.sseq = o.lss.seqs[0]
-					} else {
-						// If no mapping info just set to last.
-						o.sseq = state.LastSeq
+					filters = append(filters, o.cfg.FilterSubject)
+				} else {
+					for _, filter := range o.subjf {
+						filters = append(filters, filter.subject)
 					}
-					return
 				}
-				lss := &lastSeqSkipList{
-					resume: state.LastSeq,
-				}
-				for _, filter := range o.subjf {
-					if mss := o.mset.store.SubjectsState(filter.subject); len(mss) > 0 {
-						lss.seqs = append(lss.seqs, createLastSeqSkipList(mss)...)
+				for _, filter := range filters {
+					for subj := range o.mset.store.SubjectsTotals(filter) {
+						if sm, err := o.mset.store.LoadLastMsg(subj, &smv); err == nil {
+							lss.seqs = append(lss.seqs, sm.seq)
+						}
 					}
 				}
+				// Sort the skip list if needed.
+				if len(lss.seqs) > 1 {
+					sort.Slice(lss.seqs, func(i, j int) bool {
+						return lss.seqs[j] > lss.seqs[i]
+					})
+				}
 				if len(lss.seqs) == 0 {
 					o.sseq = state.LastSeq
+				} else {
+					o.sseq = lss.seqs[0]
 				}
-				// Sort the skip list
-				sort.Slice(lss.seqs, func(i, j int) bool {
-					return lss.seqs[j] > lss.seqs[i]
-				})
+				// Assign skip list.
 				o.lss = lss
-				if len(o.lss.seqs) != 0 {
-					o.sseq = o.lss.seqs[0]
-				}
 			} else if o.cfg.OptStartTime != nil {
 				// If we are here we are time based.
 				// TODO(dlc) - Once clustered can't rely on this.

server/filestore.go

@@ -2453,8 +2453,12 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 		}
 
 		mb.mu.Lock()
-		// Make sure we have fss loaded.
-		mb.ensurePerSubjectInfoLoaded()
+		var shouldExpire bool
+		if mb.cacheNotLoaded() {
+			// Make sure we have fss loaded.
+			mb.loadMsgsWithLock()
+			shouldExpire = true
+		}
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				if ss.firstNeedsUpdate {
@@ -2470,6 +2474,10 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 				}
 			}
 		}
+		if shouldExpire {
+			// Expire this cache before moving on.
+			mb.tryForceExpireCacheLocked()
+		}
 		mb.mu.Unlock()
 
 		if mb == stop {
@@ -6748,7 +6756,7 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return err
 		}
-		// indexCaceheBuf can produce fss now, so if non-nil we are good.
+		// indexCacheBuf can produce fss now, so if non-nil we are good.
 		if mb.fss != nil {
 			return nil
 		}