Repository: nats-io/nats-server
Pull request #5374 (2024-05-01 11:24:28)

Add a exprvar monitoring endpoint
<!-- Please make sure to read CONTRIBUTING.md, then delete this notice and replace it with your PR description. The below sign-off certifies that the contribution is your original work and that you license the work to the project under the Apache-2.0 license. We cannot accept contributions without it. -->

Signed-off-by: R.I.Pienaar <rip@devco.net>



Patch:
diff --git a/server/events.go b/server/events.go
index 40c2d6d1..aefbdbd8 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1121,6 +1121,10 @@ func (s *Server) initEventTracking() {
 			optz := &ProfilezEventOptions{}
 			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.profilez(&optz.ProfilezOptions), nil })
 		},
+		"EXPVARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+			optz := &ExpvarzEventOptions{}
+			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.expvarz(optz), nil })
+		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
@@ -1789,6 +1793,11 @@ type ProfilezEventOptions struct {
 	EventFilterOptions
 }
 
+// In the context of system events, ExpvarzEventOptions are options passed to Expvarz
+type ExpvarzEventOptions struct {
+	EventFilterOptions
+}
+
 // returns true if the request does NOT apply to this server and can be ignored.
 // DO NOT hold the server lock when
 func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
diff --git a/server/filestore.go b/server/filestore.go
index 36a0e356..6c69c818 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -52,8 +52,6 @@ type FileStoreConfig struct {
 	BlockSize uint64
 	// CacheExpire is how long with no activity until we expire the cache.
 	CacheExpire time.Duration
-	// SubjectStateExpire is how long with no activity until we expire a msg block's subject state.
-	SubjectStateExpire time.Duration
 	// SyncInterval is how often we sync to disk in the background.
 	SyncInterval time.Duration
 	// SyncAlways is when the stream should sync all data writes.
@@ -219,13 +217,11 @@ type msgBlock struct {
 	lwts       int64
 	llts       int64
 	lrts       int64
-	lsts       int64
 	llseq      uint64
 	hh         hash.Hash64
 	cache      *cache
 	cloads     uint64
 	cexp       time.Duration
-	fexp       time.Duration
 	ctmr       *time.Timer
 	werr       error
 	dmap       avl.SequenceSet
@@ -300,8 +296,8 @@ const (
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
 	closeFDsIdle = 30 * time.Second
-	// default expiration time for mb.fss when idle.
-	defaultFssExpiration = 10 * time.Second
+	// maximum flush time for mb.fss when idle.
+	maxFssFlushInterval = 10 * time.Second
 	// coalesceMinimum
 	coalesceMinimum = 16 * 1024
 	// maxFlushWait is maximum we will wait to gather messages to flush.
@@ -365,9 +361,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	if fcfg.CacheExpire == 0 {
 		fcfg.CacheExpire = defaultCacheBufferExpiration
 	}
-	if fcfg.SubjectStateExpire == 0 {
-		fcfg.SubjectStateExpire = defaultFssExpiration
-	}
 	if fcfg.SyncInterval == 0 {
 		fcfg.SyncInterval = defaultSyncInterval
 	}
@@ -886,14 +879,7 @@ func (fs *fileStore) noTrackSubjects() bool {
 
 // Will init the basics for a message block.
 func (fs *fileStore) initMsgBlock(index uint32) *msgBlock {
-	mb := &msgBlock{
-		fs:         fs,
-		index:      index,
-		cexp:       fs.fcfg.CacheExpire,
-		fexp:       fs.fcfg.SubjectStateExpire,
-		noTrack:    fs.noTrackSubjects(),
-		syncAlways: fs.fcfg.SyncAlways,
-	}
+	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire, noTrack: fs.noTrackSubjects(), syncAlways: fs.fcfg.SyncAlways}
 
 	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = filepath.Join(mdir, fmt.Sprintf(blkScan, index))
@@ -2310,8 +2296,6 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		mb.loadMsgsWithLock()
 		didLoad = true
 	}
-	// Mark fss activity.
-	mb.lsts = time.Now().UnixNano()
 
 	// If we only have 1 subject currently and it matches our filter we can also set isAll.
 	if !isAll && len(mb.fss) == 1 {
@@ -2670,8 +2654,6 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 			mb.loadMsgsWithLock()
 			shouldExpire = true
 		}
-		// Mark fss activity.
-		mb.lsts = time.Now().UnixNano()
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				if ss.firstNeedsUpdate {
@@ -2964,9 +2946,6 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 				mb.loadMsgsWithLock()
 				shouldExpire = true
 			}
-			// Mark fss activity.
-			mb.lsts = time.Now().UnixNano()
-
 			var havePartial bool
 			for subj, ss := range mb.fss {
 				if isMatch(subj) {
@@ -2982,7 +2961,6 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					}
 				}
 			}
-
 			// See if we need to scan msgs here.
 			if havePartial {
 				// Make sure we have the cache loaded.
@@ -3056,9 +3034,6 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					mb.loadMsgsWithLock()
 					shouldExpire = true
 				}
-				// Mark fss activity.
-				mb.lsts = time.Now().UnixNano()
-
 				for subj, ss := range mb.fss {
 					if isMatch(subj) {
 						adjust += ss.Msgs
@@ -3626,9 +3601,6 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 			}
 			shouldExpire = true
 		}
-		// Mark fss activity.
-		mb.lsts = time.Now().UnixNano()
-
 		if ss := mb.fss[subj]; ss != nil {
 			// Adjust first if it was not where we thought it should be.
 			if i != start {
@@ -4573,16 +4545,9 @@ func (mb *msgBlock) clearCacheAndOffset() {
 
 // Lock should be held.
 func (mb *msgBlock) clearCache() {
-	if mb.ctmr != nil {
-		tsla := mb.sinceLastActivity()
-		if mb.fss == nil || tsla > mb.fexp {
-			// Force
-			mb.fss = nil
-			mb.ctmr.Stop()
-			mb.ctmr = nil
-		} else {
-			mb.resetCacheExpireTimer(mb.fexp - tsla)
-		}
+	if mb.ctmr != nil && mb.fss == nil {
+		mb.ctmr.Stop()
+		mb.ctmr = nil
 	}
 
 	if mb.cache == nil {
@@ -4647,7 +4612,7 @@ func (mb *msgBlock) tryExpireWriteCache() []byte {
 
 // Lock should be held.
 func (mb *msgBlock) expireCacheLocked() {
-	if mb.cache == nil && mb.fss == nil {
+	if mb.cache == nil {
 		if mb.ctmr != nil {
 			mb.ctmr.Stop()
 			mb.ctmr = nil
@@ -4855,8 +4820,6 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
 			return err
 		}
-		// Mark fss activity.
-		mb.lsts = time.Now().UnixNano()
 		if ss := mb.fss[subj]; ss != nil {
 			ss.Msgs++
 			ss.Last = seq
@@ -5257,6 +5220,10 @@ func (fs *fileStore) syncBlocks() {
 	}
 	blks := append([]*msgBlock(nil), fs.blks...)
 	lmb := fs.lmb
+	fssFlushInterval := fs.fcfg.SyncInterval
+	if fssFlushInterval > maxFssFlushInterval {
+		fssFlushInterval = maxFssFlushInterval
+	}
 	fs.mu.RUnlock()
 
 	var markDirty bool
@@ -5271,6 +5238,11 @@ func (fs *fileStore) syncBlocks() {
 		if mb.mfd != nil && mb.sinceLastWriteActivity() > closeFDsIdle {
 			mb.dirtyCloseWithRemove(false)
 		}
+		// Check our fss subject metadata.
+		// If we have no activity within sync interval remove.
+		if mb.fssLoaded() && mb.sinceLastActivity() > fssFlushInterval {
+			mb.fss = nil
+		}
 
 		// Check if we should compact here as well.
 		// Do not compact last mb.
@@ -5465,8 +5437,6 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		mb.fss = make(map[string]*SimpleState)
 		popFss = true
 	}
-	// Mark fss activity.
-	mb.lsts = time.Now().UnixNano()
 
 	lbuf := uint32(len(buf))
 	var seq uint64
@@ -5739,6 +5709,12 @@ func (mb *msgBlock) fssNotLoaded() bool {
 	return mb.fss == nil && !mb.noTrack
 }
 
+// Report if we have our fss loaded.
+// Lock should be held.
+func (mb *msgBlock) fssLoaded() bool {
+	return mb.fss != nil
+}
+
 // Wrap openBlock for the gated semaphore processing.
 // Lock should be held
 func (mb *msgBlock) openBlock() (*os.File, error) {
@@ -6231,9 +6207,6 @@ func (fs *fileStore) loadLast(subj string, sm *StoreMsg) (lsm *StoreMsg, err err
 			mb.mu.Unlock()
 			return nil, err
 		}
-		// Mark fss activity.
-		mb.lsts = time.Now().UnixNano()
-
 		var l uint64
 		// Optimize if subject is not a wildcard.
 		if !wc {
@@ -6482,9 +6455,6 @@ func (mb *msgBlock) sinceLastActivity() time.Duration {
 	if mb.llts > last {
 		last = mb.llts
 	}
-	if mb.lsts > last {
-		last = mb.lsts
-	}
 	return time.Since(time.Unix(0, last).UTC())
 }
 
@@ -7467,8 +7437,6 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 	if len(mb.fss) > 0 {
 		// Make sure we run the cache expire timer.
 		mb.llts = time.Now().UnixNano()
-		// Mark fss activity.
-		mb.lsts = time.Now().UnixNano()
 		mb.startCacheExpireTimer()
 	}
 	return nil
@@ -7478,10 +7446,6 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 // Lock should be held
 func (mb *msgBlock) ensurePerSubjectInfoLoaded() error {
 	if mb.fss != nil || mb.noTrack {
-		if mb.fss != nil {
-			// Mark fss activity.
-			mb.lsts = time.Now().UnixNano()
-		}
 		return nil
 	}
 	if mb.msgs == 0 {
@@ -7644,7 +7608,9 @@ const (
 	fullStateVersion = uint8(1)
 )
 
-// This go routine periodically writes out our full stream state index.
+// This go routine runs and receives kicks to write out our full stream state index.
+// This will get kicked when we create a new block or when we delete a block in general.
+// This is also called during Stop().
 func (fs *fileStore) flushStreamStateLoop(qch, done chan struct{}) {
 	// Signal we are done on exit.
 	defer close(done)
@@ -7856,24 +7822,36 @@ func (fs *fileStore) _writeFullState(force bool) error {
 		fs.debug("WriteFullState reallocated from %d to %d", sz, cap(buf))
 	}
 
-	// Only warn about construction time since file write not holding any locks.
-	if took := time.Since(start); took > time.Minute {
-		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
+	// Write to a tmp file and rename.
+	const tmpPre = streamStreamStateFile + tsep
+	f, err := os.CreateTemp(filepath.Join(fs.fcfg.StoreDir, msgDir), tmpPre)
+	if err != nil {
+		return err
+	}
+	tmpName := f.Name()
+	defer os.Remove(tmpName)
+	if _, err = f.Write(buf); err == nil && fs.fcfg.SyncAlways {
+		f.Sync()
+	}
+	f.Close()
+	if err != nil {
+		return err
 	}
 
-	// Write our update index.db
-	// Protect with dios.
-	<-dios
-	err := os.WriteFile(fn, buf, defaultFilePerms)
-	dios <- struct{}{}
-
-	// Update dirty if successful.
-	if err == nil {
-		fs.mu.Lock()
+	// Rename into position under our lock, clear prior dirty pending on success.
+	fs.mu.Lock()
+	if !fs.closed {
+		if err := os.Rename(tmpName, fn); err != nil {
+			fs.mu.Unlock()
+			return err
+		}
 		fs.dirty -= priorDirty
-		fs.mu.Unlock()
 	}
+	fs.mu.Unlock()
 
+	if took := time.Since(start); took > time.Minute {
+		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
+	}
 	return nil
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index 3ff91cbe..6a5653d0 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -20,6 +20,7 @@ import (
 	"crypto/x509"
 	"encoding/hex"
 	"encoding/json"
+	"expvar"
 	"fmt"
 	"net"
 	"net/http"
@@ -2202,9 +2203,9 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 	}
 	s.mu.Unlock()
 
-	var leafnodes []*LeafInfo
+	leafnodes := make([]*LeafInfo, 0, len(lconns))
+
 	if len(lconns) > 0 {
-		leafnodes = make([]*LeafInfo, 0, len(lconns))
 		for _, ln := range lconns {
 			ln.mu.Lock()
 			lni := &LeafInfo{
@@ -2231,6 +2232,7 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 			leafnodes = append(leafnodes, lni)
 		}
 	}
+
 	return &Leafz{
 		ID:       s.ID(),
 		Now:      time.Now().UTC(),
@@ -3671,6 +3673,30 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	return health
 }
 
+type ExpvarzStatus struct {
+	Memstats json.RawMessage `json:"memstats"`
+	Cmdline  json.RawMessage `json:"cmdline"`
+}
+
+func (s *Server) expvarz(_ *ExpvarzEventOptions) *ExpvarzStatus {
+	var stat ExpvarzStatus
+
+	const memStatsKey = "memstats"
+	const cmdLineKey = "cmdline"
+
+	expvar.Do(func(v expvar.KeyValue) {
+		switch v.Key {
+		case memStatsKey:
+			stat.Memstats = json.RawMessage(v.Value.String())
+
+		case cmdLineKey:
+			stat.Cmdline = json.RawMessage(v.Value.String())
+		}
+	})
+
+	return &stat
+}
+
 type ProfilezStatus struct {
 	Profile []byte `json:"profile"`
 	Error   string `json:"error"`
diff --git a/server/stream.go b/server/stream.go
index b6632911..11e4952d 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2469,9 +2469,6 @@ func (mset *stream) scheduleSetupMirrorConsumerRetry() {
 	})
 }
 
-// How long we wait for a response from a consumer create request for a source or mirror.
-var srcConsumerWaitTime = 30 * time.Second
-
 // Setup our mirror consumer.
 // Lock should be held.
 func (mset *stream) setupMirrorConsumer() error {
@@ -2730,9 +2727,9 @@ func (mset *stream) setupMirrorConsumer() error {
 			}
 			mset.mu.Unlock()
 			ready.Wait()
-		case <-time.After(srcConsumerWaitTime):
+		case <-time.After(5 * time.Second):
 			mset.unsubscribe(crSub)
-			// We already waited 30 seconds, let's retry now.
+			// We already waited 5 seconds, let's retry now.
 			retry = true
 		}
 	}()
@@ -3087,9 +3084,9 @@ func (mset *stream) setupSourceConsumer(iname string, seq uint64, startTime time
 			}
 			mset.mu.Unlock()
 			ready.Wait()
-		case <-time.After(srcConsumerWaitTime):
+		case <-time.After(5 * time.Second):
 			mset.unsubscribe(crSub)
-			// We already waited 30 seconds, let's retry now.
+			// We already waited 5 seconds, let's retry now.
 			retry = true
 		}
 	}()
@@ -3300,27 +3297,15 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 			mset.mu.RLock()
 			accName, sname, iname := mset.acc.Name, mset.cfg.Name, si.iname
 			mset.mu.RUnlock()
-
-			// Can happen temporarily all the time during normal operations when the sourcing stream
-			// is working queue/interest with a limit and discard new.
-			// TODO - Improve sourcing to WQ with limit and new to use flow control rather than re-creating the consumer.
-			if errors.Is(err, ErrMaxMsgs) {
-				// Do not need to do a full retry that includes finding the last sequence in the stream
-				// for that source. Just re-create starting with the seq we couldn't store instead.
-				mset.mu.Lock()
-				mset.retrySourceConsumerAtSeq(iname, si.sseq)
-				mset.mu.Unlock()
-			} else {
-				// Log some warning for errors other than errLastSeqMismatch or errMaxMsgs.
-				if !errors.Is(err, errLastSeqMismatch) {
-					s.RateLimitWarnf("Error processing inbound source %q for '%s' > '%s': %v",
-						iname, accName, sname, err)
-				}
-				// Retry in all type of errors.
-				// This will make sure the source is still in mset.sources map,
-				// find the last sequence and then call setupSourceConsumer.
-				mset.retrySourceConsumer(iname)
+			// Log some warning for errors other than errLastSeqMismatch
+			if err != errLastSeqMismatch {
+				s.RateLimitWarnf("Error processing inbound source %q for '%s' > '%s': %v",
+					iname, accName, sname, err)
 			}
+			// Retry in all type of errors.
+			// This will make sure the source is still in mset.sources map,
+			// find the last sequence and then call setupSourceConsumer.
+			mset.retrySourceConsumer(iname)
 		}
 		return false
 	}


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 400b9d03..7e6e7b9e 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1674,8 +1674,8 @@ func TestSystemAccountWithGateways(t *testing.T) {
 	nca.Flush()
 
 	// If this tests fails with wrong number after 10 seconds we may have
-	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 56, sa)
+	// added a new initial subscription for the eventing system.
+	checkExpectedSubs(t, 58, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
@@ -2990,6 +2990,7 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 		{"HEALTHZ", nil, &JSzOptions{}, []string{"status"}},
 		{"HEALTHZ", &HealthzOptions{JSEnabledOnly: true}, &JSzOptions{}, []string{"status"}},
 		{"HEALTHZ", &HealthzOptions{JSServerOnly: true}, &JSzOptions{}, []string{"status"}},
+		{"EXPVARZ", nil, &ExpvarzStatus{}, []string{"memstats", "cmdline"}},
 	}
 
 	for i, test := range tests {
@@ -3058,8 +3059,10 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 					strings.ToLower(test.endpoint), string(msg.Data))
 			} else {
 				for _, respField := range test.respField {
-					if _, ok := resp[respField]; !ok {
+					if val, ok := resp[respField]; !ok {
 						t.Fatalf("Error finding: %s in %s", respField, resp)
+					} else if val == nil {
+						t.Fatalf("Nil value found: %s in %s", respField, resp)
 					}
 				}
 			}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index fe89cdb9..e84ef74d 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3793,8 +3793,8 @@ func (fs *fileStore) reportMeta() (hasPSIM, hasAnyFSS bool) {
 func TestFileStoreExpireSubjectMeta(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		fcfg.BlockSize = 1024
-		fcfg.CacheExpire = 500 * time.Millisecond
-		fcfg.SubjectStateExpire = time.Second
+		fcfg.CacheExpire = time.Second
+		fcfg.SyncInterval = time.Second
 		cfg := StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 1}
 		fs, err := newFileStoreWithCreated(fcfg, cfg, time.Now(), prf(&fcfg), nil)
 		require_NoError(t, err)
@@ -3821,7 +3821,7 @@ func TestFileStoreExpireSubjectMeta(t *testing.T) {
 		}
 
 		// Make sure we clear mb fss meta
-		checkFor(t, fcfg.SubjectStateExpire*2, 500*time.Millisecond, func() error {
+		checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
 			if _, hasAnyFSS := fs.reportMeta(); hasAnyFSS {
 				return fmt.Errorf("Still have mb fss state")
 			}
@@ -3832,7 +3832,7 @@ func TestFileStoreExpireSubjectMeta(t *testing.T) {
 		_, err = fs.LoadLastMsg("kv.22", nil)
 		require_NoError(t, err)
 		// Make sure we clear mb fss meta
-		checkFor(t, fcfg.SubjectStateExpire*2, 500*time.Millisecond, func() error {
+		checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
 			if _, hasAnyFSS := fs.reportMeta(); hasAnyFSS {
 				return fmt.Errorf("Still have mb fss state")
 			}
@@ -3923,7 +3923,7 @@ func TestFileStoreSubjectStateCacheExpiration(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		fcfg.BlockSize = 32
 		fcfg.CacheExpire = time.Second
-		fcfg.SubjectStateExpire = time.Second
+		fcfg.SyncInterval = time.Second
 		cfg := StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 2}
 		fs, err := newFileStoreWithCreated(fcfg, cfg, time.Now(), prf(&fcfg), nil)
 		require_NoError(t, err)
@@ -6336,7 +6336,7 @@ func TestFileStorePurgeExBufPool(t *testing.T) {
 func TestFileStoreFSSMeta(t *testing.T) {
 	sd := t.TempDir()
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: sd, BlockSize: 100, CacheExpire: 200 * time.Millisecond, SubjectStateExpire: time.Second},
+		FileStoreConfig{StoreDir: sd, BlockSize: 100, CacheExpire: 200 * time.Millisecond, SyncInterval: time.Second},
 		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -6358,22 +6358,23 @@ func TestFileStoreFSSMeta(t *testing.T) {
 	require_NoError(t, err)
 	require_Equal(t, p, 2)
 
-	// Make sure cache is not loaded.
-	var stillHasCache bool
+	// Make sure cache is not loaded but fss state still is.
+	var stillHasCache, noFSS bool
 	fs.mu.RLock()
 	for _, mb := range fs.blks {
 		mb.mu.RLock()
 		stillHasCache = stillHasCache || mb.cacheAlreadyLoaded()
+		noFSS = noFSS || mb.fssNotLoaded()
 		mb.mu.RUnlock()
 	}
 	fs.mu.RUnlock()
 
 	require_False(t, stillHasCache)
+	require_False(t, noFSS)
 
-	// Let fss expire via SubjectStateExpire.
+	// Let fss expire via syncInterval.
 	time.Sleep(time.Second)
 
-	var noFSS bool
 	fs.mu.RLock()
 	for _, mb := range fs.blks {
 		mb.mu.RLock()
@@ -6788,9 +6789,9 @@ func TestFileStoreWriteFullStateAfterPurgeEx(t *testing.T) {
 	require_Equal(t, ss.LastSeq, 10)
 }
 
-func TestFileStoreFSSExpire(t *testing.T) {
+func TestFileStoreMB_FSS_Expire(t *testing.T) {
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SubjectStateExpire: time.Second},
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SyncInterval: 2 * time.Second},
 		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, MaxMsgsPer: 1, Storage: FileStorage})
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -6825,78 +6826,6 @@ func TestFileStoreFSSExpire(t *testing.T) {
 	require_True(t, cache != nil)
 }
 
-func TestFileStoreFSSExpireNumPending(t *testing.T) {
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SubjectStateExpire: 2 * time.Second},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, MaxMsgsPer: 1, Storage: FileStorage})
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	msg := []byte("abc")
-	for i := 1; i <= 100_000; i++ {
-		fs.StoreMsg(fmt.Sprintf("foo.A.%d", i), nil, msg)
-		fs.StoreMsg(fmt.Sprintf("foo.B.%d", i), nil, msg)
-	}
-	// Flush fss by hand, cache should be flushed as well.
-	fs.mu.RLock()
-	for _, mb := range fs.blks {
-		mb.mu.Lock()
-		mb.fss = nil
-		mb.mu.Unlock()
-	}
-	fs.mu.RUnlock()
-
-	nb := fs.numMsgBlocks()
-	// Now execute NumPending() such that we load lots of blocks and make sure fss do not expire.
-	start := time.Now()
-	n, _ := fs.NumPending(100_000, "foo.A.*", false)
-	elapsed := time.Since(start)
-
-	require_Equal(t, n, 50_000)
-	// Make sure we did not force expire the fss. We would have loaded first half of blocks.
-	var noFss bool
-	last := nb/2 - 1
-	fs.mu.RLock()
-	for i, mb := range fs.blks {
-		mb.mu.RLock()
-		noFss = mb.fss == nil
-		mb.mu.RUnlock()
-		if noFss || i == last {
-			break
-		}
-	}
-	fs.mu.RUnlock()
-	require_False(t, noFss)
-
-	// Run again, make sure faster. This is consequence of fss being loaded now.
-	start = time.Now()
-	fs.NumPending(100_000, "foo.A.*", false)
-	require_True(t, elapsed > 2*time.Since(start))
-
-	// Now do with start past the mid-point.
-	start = time.Now()
-	fs.NumPending(150_000, "foo.B.*", false)
-	elapsed = time.Since(start)
-	time.Sleep(time.Second)
-	start = time.Now()
-	fs.NumPending(150_000, "foo.B.*", false)
-	require_True(t, elapsed > time.Since(start))
-
-	// Sleep enough so that all mb.fss should expire, which is 2s above.
-	time.Sleep(3 * time.Second)
-	fs.mu.RLock()
-	for i, mb := range fs.blks {
-		mb.mu.RLock()
-		fss := mb.fss
-		mb.mu.RUnlock()
-		if fss != nil {
-			fs.mu.RUnlock()
-			t.Fatalf("Detected loaded fss for mb %d", i)
-		}
-	}
-	fs.mu.RUnlock()
-}
-
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 578e54a8..31353741 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5373,10 +5373,6 @@ func TestJetStreamClusterLeaderStepdown(t *testing.T) {
 }
 
 func TestJetStreamClusterSourcesFilteringAndUpdating(t *testing.T) {
-	owt := srcConsumerWaitTime
-	srcConsumerWaitTime = 2 * time.Second
-	defer func() { srcConsumerWaitTime = owt }()
-
 	c := createJetStreamClusterExplicit(t, "MSR", 5)
 	defer c.shutdown()
 
@@ -5593,10 +5589,6 @@ func TestJetStreamClusterSourcesUpdateOriginError(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
-	owt := srcConsumerWaitTime
-	srcConsumerWaitTime = 2 * time.Second
-	defer func() { srcConsumerWaitTime = owt }()
-
 	test := func(t *testing.T, mirror bool, filter bool) {
 		c := createJetStreamClusterExplicit(t, "MSR", 5)
 		defer c.shutdown()
@@ -5651,7 +5643,7 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 
 		checkSync := func(msgsTest, msgsM uint64) {
 			t.Helper()
-			checkFor(t, 40*time.Second, time.Second, func() error {
+			checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
 				if tsi, err := js.StreamInfo("TEST"); err != nil {
 					return err
 				} else if msi, err := js.StreamInfo("M"); err != nil {
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index 1e322744..2d5b76d3 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -5317,10 +5317,6 @@ func TestJetStreamClusterMirrorSourceLoop(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorDeDupWindow(t *testing.T) {
-	owt := srcConsumerWaitTime
-	srcConsumerWaitTime = 2 * time.Second
-	defer func() { srcConsumerWaitTime = owt }()
-
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 308bf645..2983c9b9 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -24,7 +24,6 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -242,7 +241,7 @@ func TestJetStreamClusterSourceWorkingQueueWithLimit(t *testing.T) {
 
 	sendBatch := func(subject string, n int) {
 		for i := 0; i < n; i++ {
-			_, err = js.Publish(subject, []byte(strconv.Itoa(i)))
+			_, err = js.Publish(subject, []byte("OK"))
 			require_NoError(t, err)
 		}
 	}
@@ -268,9 +267,6 @@ func TestJetStreamClusterSourceWorkingQueueWithLimit(t *testing.T) {
 		for i := 0; i < 300; i++ {
 			m, err := ss.Fetch(1, nats.MaxWait(3*time.Second))
 			require_NoError(t, err)
-			p, err := strconv.Atoi(string(m[0].Data))
-			require_NoError(t, err)
-			require_Equal(t, p, i)
 			time.Sleep(11 * time.Millisecond)
 			err = m[0].Ack()
 			require_NoError(t, err)
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 9947db65..949fdb8e 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -4044,7 +4044,7 @@ func TestMonitorAccountz(t *testing.T) {
 	body = string(readBody(t, fmt.Sprintf("http://127.0.0.1:%d%s?acc=$SYS", s.MonitorAddr().Port, AccountzPath)))
 	require_Contains(t, body, `"account_detail": {`)
 	require_Contains(t, body, `"account_name": "$SYS",`)
-	require_Contains(t, body, `"subscriptions": 50,`)
+	require_Contains(t, body, `"subscriptions": 52,`)
 	require_Contains(t, body, `"is_system": true,`)
 	require_Contains(t, body, `"system_account": "$SYS"`)
 
@@ -4111,7 +4111,7 @@ func TestMonitorAccountzOperatorMode(t *testing.T) {
 	body = string(readBody(t, fmt.Sprintf("http://127.0.0.1:%d%s?acc=%s", s.MonitorAddr().Port, AccountzPath, sysPub)))
 	require_Contains(t, body, `"account_detail": {`)
 	require_Contains(t, body, fmt.Sprintf(`"account_name": "%s",`, sysPub))
-	require_Contains(t, body, `"subscriptions": 50,`)
+	require_Contains(t, body, `"subscriptions": 52,`)
 	require_Contains(t, body, `"is_system": true,`)
 	require_Contains(t, body, fmt.Sprintf(`"system_account": "%s"`, sysPub))
 
diff --git a/server/norace_test.go b/server/norace_test.go
index 9d61625e..14ae42a1 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1767,10 +1767,6 @@ func TestNoRaceJetStreamSuperClusterMixedModeMirrors(t *testing.T) {
 }
 
 func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
-	owt := srcConsumerWaitTime
-	srcConsumerWaitTime = 2 * time.Second
-	defer func() { srcConsumerWaitTime = owt }()
-
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
 
@@ -9270,10 +9266,6 @@ func TestNoRaceJetStreamAPIDispatchQueuePending(t *testing.T) {
 }
 
 func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
-	owt := srcConsumerWaitTime
-	srcConsumerWaitTime = 2 * time.Second
-	defer func() { srcConsumerWaitTime = owt }()
-
 	// Check calculations first.
 	for i := 1; i <= 20; i++ {
 		backoff := calculateRetryBackoff(i)
@@ -9330,8 +9322,8 @@ func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
 	sldr := c.streamLeader(globalAccountName, "TEST")
 	sldr.Shutdown()
 
-	// Wait for just greater than 5s. We should only see 1 request during this time.
-	time.Sleep(6 * time.Second)
+	// Wait for just greater than 10s. We should only see 1 request during this time.
+	time.Sleep(11 * time.Second)
 	// There should have been 2 requests, one for mirror, one for source
 	n, _, _ := sub.Pending()
 	require_Equal(t, n, 2)