Repository: nats-io/nats-server
Pull request #5377 (2024-05-01 19:33:53)

[IMPROVED] Updated subject state expiration 
Updated subject state expiration to be handled by individual block cache expiration logic.
Also added in additional filestore config to control this if needed, still defaults to 10s.

Signed-off-by: Derek Collison <derek@nats.io>


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 6c69c818..1db4360c 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -52,6 +52,8 @@ 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.
@@ -222,6 +224,7 @@ type msgBlock struct {
 	cache      *cache
 	cloads     uint64
 	cexp       time.Duration
+	fexp       time.Duration
 	ctmr       *time.Timer
 	werr       error
 	dmap       avl.SequenceSet
@@ -296,8 +299,8 @@ const (
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
 	closeFDsIdle = 30 * time.Second
-	// maximum flush time for mb.fss when idle.
-	maxFssFlushInterval = 10 * time.Second
+	// default expiration time for mb.fss when idle.
+	defaultFssExpiration = 10 * time.Second
 	// coalesceMinimum
 	coalesceMinimum = 16 * 1024
 	// maxFlushWait is maximum we will wait to gather messages to flush.
@@ -361,6 +364,9 @@ 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
 	}
@@ -879,7 +885,14 @@ 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, noTrack: fs.noTrackSubjects(), syncAlways: fs.fcfg.SyncAlways}
+	mb := &msgBlock{
+		fs:         fs,
+		index:      index,
+		cexp:       fs.fcfg.CacheExpire,
+		fexp:       fs.fcfg.SubjectStateExpire,
+		noTrack:    fs.noTrackSubjects(),
+		syncAlways: fs.fcfg.SyncAlways,
+	}
 
 	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = filepath.Join(mdir, fmt.Sprintf(blkScan, index))
@@ -4545,9 +4558,16 @@ func (mb *msgBlock) clearCacheAndOffset() {
 
 // Lock should be held.
 func (mb *msgBlock) clearCache() {
-	if mb.ctmr != nil && mb.fss == nil {
-		mb.ctmr.Stop()
-		mb.ctmr = nil
+	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.cache == nil {
@@ -4612,7 +4632,7 @@ func (mb *msgBlock) tryExpireWriteCache() []byte {
 
 // Lock should be held.
 func (mb *msgBlock) expireCacheLocked() {
-	if mb.cache == nil {
+	if mb.cache == nil && mb.fss == nil {
 		if mb.ctmr != nil {
 			mb.ctmr.Stop()
 			mb.ctmr = nil
@@ -5220,10 +5240,6 @@ 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
@@ -5238,11 +5254,6 @@ 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.
@@ -5709,12 +5720,6 @@ 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) {
@@ -7608,9 +7613,7 @@ const (
 	fullStateVersion = uint8(1)
 )
 
-// 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().
+// This go routine periodically writes out our full stream state index.
 func (fs *fileStore) flushStreamStateLoop(qch, done chan struct{}) {
 	// Signal we are done on exit.
 	defer close(done)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index e84ef74d..ea78da83 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 = time.Second
-		fcfg.SyncInterval = time.Second
+		fcfg.CacheExpire = 500 * time.Millisecond
+		fcfg.SubjectStateExpire = 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, 10*time.Second, 500*time.Millisecond, func() error {
+		checkFor(t, fcfg.SubjectStateExpire*2, 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, 10*time.Second, 500*time.Millisecond, func() error {
+		checkFor(t, fcfg.SubjectStateExpire*2, 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.SyncInterval = time.Second
+		fcfg.SubjectStateExpire = 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, SyncInterval: time.Second},
+		FileStoreConfig{StoreDir: sd, BlockSize: 100, CacheExpire: 200 * time.Millisecond, SubjectStateExpire: time.Second},
 		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -6358,23 +6358,22 @@ func TestFileStoreFSSMeta(t *testing.T) {
 	require_NoError(t, err)
 	require_Equal(t, p, 2)
 
-	// Make sure cache is not loaded but fss state still is.
-	var stillHasCache, noFSS bool
+	// Make sure cache is not loaded.
+	var stillHasCache 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.
+	// Let fss expire via SubjectStateExpire.
 	time.Sleep(time.Second)
 
+	var noFSS bool
 	fs.mu.RLock()
 	for _, mb := range fs.blks {
 		mb.mu.RLock()