[IMPROVED] Optimizations for PurgeEx from KV PurgeDeletes() clients. (#6801)

Various optimizations and introduction of AllLastSeqs().

Filestore specific items.
1. Internal calls to fetch a msg now use a NoCopy version to avoid
unnecessary allocations.
2. We introduced a last purge time to detect repeated calls to Purge
from a client KV PurgeDeletes operation.
3. Access times for cache lifetime management are now from a single
ticker updating an atomic every 100ms.
    4. filteredPendingLocked now detects non-wildcard filters.
5. AllLastSeqs() is an optimized version to return all last sequences
for all subjects. Helpful for KV watchers.
6. Optimized MultiLastSeqs() and allow it to call into AllLastSeqs()
when appropriate.
7. Allow multiple tombstones to be written async and flushed all at
once.

Signed-off-by: Derek Collison <[email protected]>

Author: neilalexander

server/client.go

@@ -4329,7 +4329,7 @@ func sliceHeader(key string, hdr []byte) []byte {
 	if len(hdr) == 0 {
 		return nil
 	}
-	index := bytes.Index(hdr, []byte(key))
+	index := bytes.Index(hdr, stringToBytes(key))
 	hdrLen := len(hdr)
 	// Check that we have enough characters, this will handle the -1 case of the key not
 	// being found and will also handle not having enough characters for trailing CRLF.

server/consumer.go

@@ -5322,9 +5322,6 @@ func (o *consumer) selectStartingSeqNo() {
 				if mmp == 1 {
 					o.sseq = state.FirstSeq
 				} else {
-					// A threshold for when we switch from get last msg to subjects state.
-					const numSubjectsThresh = 256
-					lss := &lastSeqSkipList{resume: state.LastSeq}
 					var filters []string
 					if o.subjf == nil {
 						filters = append(filters, o.cfg.FilterSubject)
@@ -5333,24 +5330,10 @@ func (o *consumer) selectStartingSeqNo() {
 							filters = append(filters, filter.subject)
 						}
 					}
-					for _, filter := range filters {
-						if st := o.mset.store.SubjectsTotals(filter); len(st) < numSubjectsThresh {
-							var smv StoreMsg
-							for subj := range st {
-								if sm, err := o.mset.store.LoadLastMsg(subj, &smv); err == nil {
-									lss.seqs = append(lss.seqs, sm.seq)
-								}
-							}
-						} else if mss := o.mset.store.SubjectsState(filter); len(mss) > 0 {
-							for _, ss := range mss {
-								lss.seqs = append(lss.seqs, ss.Last)
-							}
-						}
-					}
-					// Sort the skip list if needed.
-					if len(lss.seqs) > 1 {
-						slices.Sort(lss.seqs)
-					}
+
+					lss := &lastSeqSkipList{resume: state.LastSeq}
+					lss.seqs, _ = o.mset.store.MultiLastSeqs(filters, 0, 0)
+
 					if len(lss.seqs) == 0 {
 						o.sseq = state.LastSeq
 					} else {
@@ -5874,7 +5857,10 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 
 	// Check if this message was pending.
 	p, wasPending := o.pending[sseq]
-	rdc := o.deliveryCount(sseq)
+	var rdc uint64
+	if wasPending {
+		rdc = o.deliveryCount(sseq)
+	}
 
 	o.mu.Unlock()