Repository: nats-io/nats-server
Pull request #4917 (2023-12-29 21:34:53)

Cherry-pick PRs into v2.10.8 release branch
Includes: 

- https://github.com/nats-io/nats-server/pull/4916
- https://github.com/nats-io/nats-server/pull/4915



Patch:
diff --git a/server/certidp/certidp.go b/server/certidp/certidp.go
index f7b660df..e066ed43 100644
--- a/server/certidp/certidp.go
+++ b/server/certidp/certidp.go
@@ -52,11 +52,24 @@ var (
 	}
 )
 
+// GetStatusAssertionStr returns the corresponding string representation of the StatusAssertion.
 func GetStatusAssertionStr(sa int) string {
-	return StatusAssertionValToStr[StatusAssertionIntToVal[sa]]
+	// If the provided status assertion value is not found in the map (StatusAssertionIntToVal),
+	// the function defaults to "unknown" to avoid defaulting to "good," which is the default iota value
+	// for the ocsp.StatusAssertion enumeration (https://pkg.go.dev/golang.org/x/crypto/ocsp#pkg-constants).
+	// This ensures that we don't unintentionally default to "good" when there's no map entry.
+	v, ok := StatusAssertionIntToVal[sa]
+	if !ok {
+		// set unknown as fallback
+		v = ocsp.Unknown
+	}
+
+	return StatusAssertionValToStr[v]
 }
 
 func (sa StatusAssertion) MarshalJSON() ([]byte, error) {
+	// This ensures that we don't unintentionally default to "good" when there's no map entry.
+	// (see more details in the GetStatusAssertionStr() comment)
 	str, ok := StatusAssertionValToStr[sa]
 	if !ok {
 		// set unknown as fallback
@@ -66,6 +79,8 @@ func (sa StatusAssertion) MarshalJSON() ([]byte, error) {
 }
 
 func (sa *StatusAssertion) UnmarshalJSON(in []byte) error {
+	// This ensures that we don't unintentionally default to "good" when there's no map entry.
+	// (see more details in the GetStatusAssertionStr() comment)
 	v, ok := StatusAssertionStrToVal[strings.ReplaceAll(string(in), "\"", "")]
 	if !ok {
 		// set unknown as fallback
diff --git a/server/filestore.go b/server/filestore.go
index e4ba028b..c7244c82 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1481,7 +1481,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 	if _, err := os.Stat(pdir); err == nil {
 		os.RemoveAll(pdir)
 	}
-
 	// Grab our stream state file and load it in.
 	fn := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)
 	buf, err := os.ReadFile(fn)
@@ -2031,6 +2030,7 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		}
 
 		// If we are here we have to process the interior messages of this blk.
+		// This will load fss as well.
 		if err := mb.loadMsgsWithLock(); err != nil {
 			mb.mu.Unlock()
 			break
@@ -2040,7 +2040,6 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		var needNextFirst bool
 
 		// Walk messages and remove if expired.
-		mb.ensurePerSubjectInfoLoaded()
 		fseq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq)
 		for seq := fseq; seq <= lseq; seq++ {
 			sm, err := mb.cacheLookup(seq, &smv)
@@ -2203,13 +2202,8 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 
 	fseq, isAll, subs := start, filter == _EMPTY_ || filter == fwcs, []string{filter}
 
-	if mb.cacheNotLoaded() {
-		if err := mb.loadMsgsWithLock(); err != nil {
-			return nil, false, err
-		}
-		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
-			return nil, false, err
-		}
+	if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
+		return nil, false, err
 	}
 
 	// If we only have 1 subject currently and it matches our filter we can also set isAll.
@@ -2257,6 +2251,15 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		return nil, false, ErrStoreMsgNotFound
 	}
 
+	var didLoad bool
+	// Need messages loaded from here on out.
+	if mb.cacheNotLoaded() {
+		if err := mb.loadMsgsWithLock(); err != nil {
+			return nil, false, err
+		}
+		didLoad = true
+	}
+
 	if sm == nil {
 		sm = new(StoreMsg)
 	}
@@ -2288,7 +2291,7 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		mb.llseq = llseq
 	}
 
-	return nil, false, ErrStoreMsgNotFound
+	return nil, didLoad, ErrStoreMsgNotFound
 }
 
 // This will traverse a message block and generate the filtered pending.
@@ -2540,7 +2543,7 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 
 		mb.mu.Lock()
 		var shouldExpire bool
-		if mb.fss == nil {
+		if mb.fssNotLoaded() {
 			// Make sure we have fss loaded.
 			mb.loadMsgsWithLock()
 			shouldExpire = true
@@ -2717,7 +2720,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 			}
 			// If we are here we need to at least scan the subject fss.
 			// Make sure we have fss loaded.
-			if mb.cacheNotLoaded() {
+			if mb.fssNotLoaded() {
 				mb.loadMsgsWithLock()
 				shouldExpire = true
 			}
@@ -2738,6 +2741,11 @@ 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.
+				if mb.cacheNotLoaded() {
+					mb.loadMsgsWithLock()
+					shouldExpire = true
+				}
 				// Clear on partial.
 				t = 0
 				var smv StoreMsg
@@ -3272,9 +3280,14 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 			continue
 		}
 		mb.mu.Lock()
-		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
-			mb.mu.Unlock()
-			return 0, err
+		var shouldExpire bool
+		if mb.fssNotLoaded() {
+			// Make sure we have fss loaded.
+			if err := mb.loadMsgsWithLock(); err != nil {
+				mb.mu.Unlock()
+				return 0, err
+			}
+			shouldExpire = true
 		}
 		if ss := mb.fss[subj]; ss != nil {
 			// Adjust first if it was not where we thought it should be.
@@ -3289,6 +3302,11 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 			mb.mu.Unlock()
 			return ss.First, nil
 		}
+		// If we did not find it and we loaded this msgBlock try to expire as long as not the last.
+		if shouldExpire {
+			// Expire this cache before moving on.
+			mb.tryForceExpireCacheLocked()
+		}
 		mb.mu.Unlock()
 	}
 	return 0, nil
@@ -3395,9 +3413,6 @@ func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
 			}
 			// Grab the ss entry for this subject in case sparse.
 			mb.mu.Lock()
-			if mb.cacheNotLoaded() {
-				mb.loadMsgsWithLock()
-			}
 			mb.ensurePerSubjectInfoLoaded()
 			ss := mb.fss[subj]
 			if ss != nil && ss.firstNeedsUpdate {
@@ -4283,7 +4298,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
@@ -4323,9 +4338,8 @@ func (mb *msgBlock) expireCacheLocked() {
 		mb.cache.wp = 0
 	}
 
-	// Check if we can clear out our fss and idx unless under force expire.
-	// We used to hold onto the idx longer but removes need buf now so no point.
-	mb.fss = nil
+	// Check if we can clear out our idx unless under force expire.
+	// fss we keep longer and expire under sync timer checks.
 	mb.clearCache()
 }
 
@@ -4714,9 +4728,7 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 	// Wait for disk I/O slots to become available. This prevents us from
 	// running away with system resources.
 	<-dios
-	defer func() {
-		dios <- struct{}{}
-	}()
+	defer func() { dios <- struct{}{} }()
 
 	alg := mb.fs.fcfg.Compression
 	mb.mu.Lock()
@@ -4888,6 +4900,7 @@ func (fs *fileStore) syncBlocks() {
 	}
 	blks := append([]*msgBlock(nil), fs.blks...)
 	lmb := fs.lmb
+	syncInterval := fs.fcfg.SyncInterval
 	fs.mu.RUnlock()
 
 	var markDirty bool
@@ -4902,6 +4915,12 @@ 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() > syncInterval {
+			mb.fss = nil
+		}
+
 		// Check if we should compact here as well.
 		// Do not compact last mb.
 		var needsCompact bool
@@ -5311,6 +5330,18 @@ func (mb *msgBlock) cacheNotLoaded() bool {
 	return !mb.cacheAlreadyLoaded()
 }
 
+// Report if our fss is not loaded.
+// Lock should be held.
+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
+}
+
 // Used to load in the block contents.
 // Lock should be held and all conditionals satisfied prior.
 func (mb *msgBlock) loadBlock(buf []byte) ([]byte, error) {
@@ -5469,11 +5500,12 @@ func (mb *msgBlock) fetchMsg(seq uint64, sm *StoreMsg) (*StoreMsg, bool, error)
 			return nil, false, err
 		}
 	}
+	llseq := mb.llseq
 	fsm, err := mb.cacheLookup(seq, sm)
 	if err != nil {
 		return nil, false, err
 	}
-	expireOk := seq == atomic.LoadUint64(&mb.last.seq) && mb.llseq == seq
+	expireOk := (seq == atomic.LoadUint64(&mb.last.seq) && llseq == seq-1) || (seq == atomic.LoadUint64(&mb.first.seq) && llseq == seq+1)
 	return fsm, expireOk, err
 }
 
@@ -5515,6 +5547,13 @@ func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 		return nil, ErrStoreMsgNotFound
 	}
 
+	// The llseq signals us when we can expire a cache at the end of a linear scan.
+	// We want to only update when we know the last reads (multiple consumers) are sequential.
+	// We want to account for forwards and backwards linear scans.
+	if mb.llseq == 0 || seq < mb.llseq || seq == mb.llseq+1 || seq == mb.llseq-1 {
+		mb.llseq = seq
+	}
+
 	// If we have a delete map check it.
 	if mb.dmap.Exists(seq) {
 		mb.llts = time.Now().UnixNano()
@@ -5537,11 +5576,6 @@ func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 
 	// Update cache activity.
 	mb.llts = time.Now().UnixNano()
-	// The llseq signals us when we can expire a cache at the end of a linear scan.
-	// We want to only update when we know the last reads (multiple consumers) are sequential.
-	if mb.llseq == 0 || seq < mb.llseq || seq == mb.llseq+1 {
-		mb.llseq = seq
-	}
 
 	li := int(bi) - mb.cache.off
 	if li >= len(mb.cache.buf) {
@@ -5610,7 +5644,7 @@ func (fs *fileStore) msgForSeq(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 		seq = fs.state.FirstSeq
 	}
 	// Make sure to snapshot here.
-	mb, lmb, lseq := fs.selectMsgBlock(seq), fs.lmb, fs.state.LastSeq
+	mb, lseq := fs.selectMsgBlock(seq), fs.state.LastSeq
 	fs.mu.RUnlock()
 
 	if mb == nil {
@@ -5628,7 +5662,7 @@ func (fs *fileStore) msgForSeq(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 
 	// We detected a linear scan and access to the last message.
 	// If we are not the last message block we can try to expire the cache.
-	if mb != lmb && expireOk {
+	if expireOk {
 		mb.tryForceExpireCache()
 	}
 
@@ -5801,12 +5835,14 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
 		for i := bi; i < len(fs.blks); i++ {
 			mb := fs.blks[i]
 			if sm, expireOk, err := mb.firstMatching(filter, wc, start, sm); err == nil {
-				if expireOk && mb != fs.lmb {
+				if expireOk {
 					mb.tryForceExpireCache()
 				}
 				return sm, sm.seq, nil
 			} else if err != ErrStoreMsgNotFound {
 				return nil, 0, err
+			} else if expireOk {
+				mb.tryForceExpireCache()
 			}
 		}
 	}
@@ -5919,6 +5955,21 @@ func fileStoreMsgSizeEstimate(slen, maxPayload int) uint64 {
 	return uint64(emptyRecordLen + slen + 4 + maxPayload)
 }
 
+// Determine time since any last activity, read/load, write or remove.
+func (mb *msgBlock) sinceLastActivity() time.Duration {
+	if mb.closed {
+		return 0
+	}
+	last := mb.lwts
+	if mb.lrts > last {
+		last = mb.lrts
+	}
+	if mb.llts > last {
+		last = mb.llts
+	}
+	return time.Since(time.Unix(0, last).UTC())
+}
+
 // Determine time since last write or remove of a message.
 // Read lock should be held.
 func (mb *msgBlock) sinceLastWriteActivity() time.Duration {
@@ -6132,11 +6183,9 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	for i := 0; i < len(fs.blks); i++ {
 		mb := fs.blks[i]
 		mb.mu.Lock()
-		var shouldExpire bool
-		if mb.cacheNotLoaded() {
-			mb.loadMsgsWithLock()
-			shouldExpire = true
-		}
+
+		// If we do not have our fss, try to expire the cache if we have no items in this block.
+		shouldExpire := mb.fssNotLoaded()
 
 		t, f, l := mb.filteredPendingLocked(subject, wc, atomic.LoadUint64(&mb.first.seq))
 		if t == 0 {
@@ -6153,6 +6202,11 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 			l = sequence - 1
 		}
 
+		if mb.cacheNotLoaded() {
+			mb.loadMsgsWithLock()
+			shouldExpire = true
+		}
+
 		for seq := f; seq <= l; seq++ {
 			if sm, _ := mb.cacheLookup(seq, &smv); sm != nil && eq(sm.subj, subject) {
 				rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
@@ -6944,6 +6998,7 @@ func (mb *msgBlock) close(sync bool) {
 		mb.ctmr = nil
 	}
 
+	// Clear fss.
 	mb.fss = nil
 
 	// Close cache
@@ -8871,5 +8926,4 @@ func (alg StoreCompression) Decompress(buf []byte) ([]byte, error) {
 	output = append(output, checksum...)
 
 	return output, reader.Close()
-
 }
diff --git a/server/raft.go b/server/raft.go
index 1180979a..6990812c 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -1052,7 +1052,7 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	sn := fmt.Sprintf(snapFileT, snap.lastTerm, snap.lastIndex)
 	sfile := filepath.Join(snapDir, sn)
 
-	if err := os.WriteFile(sfile, n.encodeSnapshot(snap), 0640); err != nil {
+	if err := os.WriteFile(sfile, n.encodeSnapshot(snap), defaultFilePerms); err != nil {
 		n.Unlock()
 		// We could set write err here, but if this is a temporary situation, too many open files etc.
 		// we want to retry and snapshots are not fatal.
@@ -3678,7 +3678,7 @@ func writePeerState(sd string, ps *peerState) error {
 	if _, err := os.Stat(psf); err != nil && !os.IsNotExist(err) {
 		return err
 	}
-	if err := os.WriteFile(psf, encodePeerState(ps), 0640); err != nil {
+	if err := os.WriteFile(psf, encodePeerState(ps), defaultFilePerms); err != nil {
 		return err
 	}
 	return nil
@@ -3778,7 +3778,7 @@ func (n *raft) fileWriter() {
 			copy(buf[0:], n.wtv)
 			n.RUnlock()
 			<-dios
-			err := os.WriteFile(tvf, buf[:], 0640)
+			err := os.WriteFile(tvf, buf[:], defaultFilePerms)
 			dios <- struct{}{}
 			if err != nil && !n.isClosed() {
 				n.setWriteErr(err)
@@ -3790,7 +3790,7 @@ func (n *raft) fileWriter() {
 			buf := copyBytes(n.wps)
 			n.RUnlock()
 			<-dios
-			err := os.WriteFile(psf, buf, 0640)
+			err := os.WriteFile(psf, buf, defaultFilePerms)
 			dios <- struct{}{}
 			if err != nil && !n.isClosed() {
 				n.setWriteErr(err)


Test patch:
diff --git a/server/certidp/certidp_test.go b/server/certidp/certidp_test.go
new file mode 100644
index 00000000..ffc31055
--- /dev/null
+++ b/server/certidp/certidp_test.go
@@ -0,0 +1,56 @@
+// Copyright 2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package certidp
+
+import "testing"
+
+// Checks the return values of the function GetStatusAssertionStr
+func TestGetStatusAssertionStr(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    int
+		expected string
+	}{
+		{
+			name:     "GoodStatus",
+			input:    0,
+			expected: "good",
+		},
+		{
+			name:     "RevokedStatus",
+			input:    1,
+			expected: "revoked",
+		},
+		{
+			name:     "UnknownStatus",
+			input:    2,
+			expected: "unknown",
+		},
+		// Invalid status assertion value.
+		{
+			name:     "InvalidStatus",
+			input:    42,
+			expected: "unknown",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := GetStatusAssertionStr(tt.input)
+			if got != tt.expected {
+				t.Errorf("Expected GetStatusAssertionStr: %v, got %v", tt.expected, got)
+			}
+		})
+	}
+}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 427e137c..6d98f4d1 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3788,6 +3788,7 @@ func TestFileStoreExpireSubjectMeta(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		fcfg.BlockSize = 1024
 		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)
@@ -3916,6 +3917,7 @@ func TestFileStoreSubjectStateCacheExpiration(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		fcfg.BlockSize = 32
 		fcfg.CacheExpire = 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)
@@ -6318,6 +6320,129 @@ func TestFileStorePurgeExBufPool(t *testing.T) {
 	require_Equal(t, loaded, 1)
 }
 
+func TestFileStoreFSSMeta(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		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()
+
+	// This yields an internal record length of 50 bytes. So 2 msgs per blk with subject len of 1, e.g. "A" or "Z".
+	msg := bytes.Repeat([]byte("Z"), 19)
+
+	// Should leave us with |A-Z| |Z-Z| |Z-Z| |Z-A|
+	fs.StoreMsg("A", nil, msg)
+	for i := 0; i < 6; i++ {
+		fs.StoreMsg("Z", nil, msg)
+	}
+	fs.StoreMsg("A", nil, msg)
+
+	// Let cache's expire before PurgeEx which will load them back in.
+	time.Sleep(250 * time.Millisecond)
+
+	p, err := fs.PurgeEx("A", 1, 0)
+	require_NoError(t, err)
+	require_Equal(t, p, 2)
+
+	// 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 syncInterval.
+	time.Sleep(time.Second)
+
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		noFSS = noFSS || mb.fssNotLoaded()
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+
+	require_True(t, noFSS)
+}
+
+func TestFileStoreExpireCacheOnLinearWalk(t *testing.T) {
+	sd := t.TempDir()
+	expire := 250 * time.Millisecond
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, CacheExpire: expire},
+		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// This yields an internal record length of 50 bytes.
+	subj, msg := "Z", bytes.Repeat([]byte("Z"), 19)
+
+	// Store 10 messages, so 5 blocks.
+	for i := 0; i < 10; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+	// Let them all expire. This way we load as we walk and can test that we expire all blocks without
+	// needing to worry about last write times blocking forced expiration.
+	time.Sleep(expire)
+
+	checkNoCache := func() {
+		t.Helper()
+		fs.mu.RLock()
+		var stillHasCache bool
+		for _, mb := range fs.blks {
+			mb.mu.RLock()
+			stillHasCache = stillHasCache || mb.cacheAlreadyLoaded()
+			mb.mu.RUnlock()
+		}
+		fs.mu.RUnlock()
+		require_False(t, stillHasCache)
+	}
+
+	// Walk forward.
+	var smv StoreMsg
+	for seq := uint64(1); seq <= 10; seq++ {
+		_, err := fs.LoadMsg(seq, &smv)
+		require_NoError(t, err)
+	}
+	checkNoCache()
+
+	// No test walking backwards. We have this scenario when we search for starting points for sourced streams.
+	// Noticed some memory bloat when we have to search many blocks looking for a source that may be closer to the
+	// beginning of the stream (infrequently updated sourced stream).
+	for seq := uint64(10); seq >= 1; seq-- {
+		_, err := fs.LoadMsg(seq, &smv)
+		require_NoError(t, err)
+	}
+	checkNoCache()
+
+	// Now make sure still expires properly on linear scans with deleted msgs.
+	// We want to make sure we track linear updates even if message deleted.
+	_, err = fs.RemoveMsg(2)
+	require_NoError(t, err)
+	_, err = fs.RemoveMsg(9)
+	require_NoError(t, err)
+
+	// Walk forward.
+	for seq := uint64(1); seq <= 10; seq++ {
+		_, err := fs.LoadMsg(seq, &smv)
+		if seq == 2 || seq == 9 {
+			require_Error(t, err, errDeletedMsg)
+		} else {
+			require_NoError(t, err)
+		}
+	}
+	checkNoCache()
+
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////