[IMPROVED] PurgeEx only loads within subject range

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Author: MauriceVanVeen

server/filestore.go

@@ -9564,10 +9564,41 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 		fs.mu.Unlock()
 		return purged, err
 	}
+
+	var start, stop uint32
+
+	// If literal subject check for presence.
+	if wc {
+		start = fs.lmb.index
+		fs.psim.Match(stringToBytes(subject), func(_ []byte, psi *psi) {
+			// Keep track of start and stop indexes for this subject.
+			if psi.fblk < start {
+				start = psi.fblk
+			}
+			if psi.lblk > stop {
+				stop = psi.lblk
+			}
+		})
+		// None matched.
+		if stop == 0 {
+			fs.mu.Unlock()
+			return purged, nil
+		}
+	} else if info, ok := fs.psim.Find(stringToBytes(subject)); ok {
+		start, stop = info.fblk, info.lblk
+	} else {
+		fs.mu.Unlock()
+		return purged, nil
+	}
+
 	// We may remove blocks as we purge, so don't range directly on fs.blks
 	// otherwise we may jump over some (see https://github.com/nats-io/nats-server/issues/3528)
 	for i := 0; i < len(fs.blks); i++ {
 		mb := fs.blks[i]
+		// Skip if not within our range for the purge subject.
+		if mb.index < start || mb.index > stop {
+			continue
+		}
 		mb.mu.Lock()
 
 		// If we do not have our fss, try to expire the cache if we have no items in this block.

server/filestore_test.go

@@ -13574,3 +13574,65 @@ func TestFileStoreNoDirectoryNotEmptyError(t *testing.T) {
 		wg.Wait()
 	}
 }
+
+func TestFileStoreDontLoadSubjectStateIfNotPurged(t *testing.T) {
+	fcfg := FileStoreConfig{StoreDir: t.TempDir()}
+	cfg := StreamConfig{Name: "zzz", Storage: FileStorage, Subjects: []string{">"}}
+	fs, err := newFileStore(fcfg, cfg)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+	require_NoError(t, err)
+
+	fmb := fs.getFirstBlock()
+	fmb.mu.Lock()
+	fmb.fss = nil
+	fmb.mu.Unlock()
+
+	// When purging a subject that doesn't exist, we shouldn't need to load subjects for any blocks.
+	purged, err := fs.PurgeEx("bar", 0, 0)
+	require_NoError(t, err)
+	require_Equal(t, purged, 0)
+
+	fmb.mu.RLock()
+	defer fmb.mu.RUnlock()
+	require_True(t, fmb.fss == nil)
+}
+
+func TestFileStoreOnlyLoadSubjectStateOnBlocksWithinInterestRange(t *testing.T) {
+	fcfg := FileStoreConfig{StoreDir: t.TempDir()}
+	cfg := StreamConfig{Name: "zzz", Storage: FileStorage, Subjects: []string{">"}}
+	fs, err := newFileStore(fcfg, cfg)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+	require_NoError(t, err)
+	_, err = fs.newMsgBlockForWrite()
+	require_NoError(t, err)
+	_, _, err = fs.StoreMsg("bar", nil, nil, 0)
+	require_NoError(t, err)
+
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		mb.fss = nil
+		mb.mu.Unlock()
+	}
+	fs.mu.RUnlock()
+
+	// When purging a subject that only exists on a subset, we should only load blocks within its range.
+	purged, err := fs.PurgeEx("bar", 0, 0)
+	require_NoError(t, err)
+	require_Equal(t, purged, 1)
+
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+	for i, mb := range fs.blks {
+		mb.mu.Lock()
+		loadedFss := mb.fss != nil
+		mb.mu.Unlock()
+		require_Equal(t, loadedFss, i == 1)
+	}
+}