Limit the number of in-memory cached blocks per store

Signed-off-by: Neil Twigg <[email protected]>

Author: neilalexander

server/filestore.go

@@ -205,6 +205,7 @@ type fileStore struct {
 	ttls        *thw.HashWheel
 	sdm         *SDMMeta
 	lpex        time.Time // Last PurgeEx call.
+	evict       *ipQueue[*msgBlock]
 }
 
 // Represents a message store block and its data.
@@ -313,6 +314,8 @@ const (
 	maxBufReuse = 2 * 1024 * 1024
 	// default cache buffer expiration
 	defaultCacheBufferExpiration = 10 * time.Second
+	// deault cache size before eviction
+	defaultCacheEvictionThreshold = 32
 	// default sync interval
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
@@ -420,6 +423,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		qch:    make(chan struct{}),
 		fsld:   make(chan struct{}),
 		srv:    fcfg.srv,
+		evict:  newIPQueue[*msgBlock](nil, _EMPTY_),
 	}
 
 	// Register with access time service.
@@ -7058,6 +7062,14 @@ checkCache:
 	if len(buf) > 0 {
 		mb.cloads++
 		mb.startCacheExpireTimer()
+
+		// If loading this block into the cache caused us to reach the eviction
+		// threshold for cached blocks, evict the oldest one.
+		if n, _ := mb.fs.evict.push(mb); n == defaultCacheEvictionThreshold {
+			if emb, ok := mb.fs.evict.popOne(); ok && emb != mb {
+				emb.tryForceExpireCache()
+			}
+		}
 	}
 
 	return nil

server/ipqueue.go

@@ -95,15 +95,17 @@ func newIPQueue[T any](s *Server, name string, opts ...ipQueueOpt[T]) *ipQueue[T
 			},
 		},
 		name: name,
-		m:    &s.ipQueues,
 		ipQueueOpts: ipQueueOpts[T]{
 			mrs: ipQueueDefaultMaxRecycleSize,
 		},
 	}
+	if s != nil {
+		q.m = &s.ipQueues
+		q.m.Store(name, q)
+	}
 	for _, o := range opts {
 		o(&q.ipQueueOpts)
 	}
-	s.ipQueues.Store(name, q)
 	return q
 }
 
@@ -279,7 +281,7 @@ func (q *ipQueue[T]) inProgress() int64 {
 // Remove this queue from the server's map of ipQueues.
 // All ipQueue operations (such as push/pop/etc..) are still possible.
 func (q *ipQueue[T]) unregister() {
-	if q == nil {
+	if q == nil || q.m == nil {
 		return
 	}
 	q.m.Delete(q.name)