Updates based on PR feedback

Signed-off-by: Derek Collison <[email protected]>

Author: derekcollison

server/filestore_test.go

@@ -3670,6 +3670,12 @@ func TestFileStorePurgeExWithSubject(t *testing.T) {
 		// Make sure we have our state file prior to Purge call.
 		fs.forceWriteFullState()
 
+		// Capture the current index.db file.
+		sfile := filepath.Join(fcfg.StoreDir, msgDir, streamStreamStateFile)
+		buf, err := os.ReadFile(sfile)
+		require_NoError(t, err)
+		require_True(t, len(buf) > 0)
+
 		// This should purge all "foo.1"
 		p, err := fs.PurgeEx("foo.1", 1, 0)
 		require_NoError(t, err)
@@ -3679,12 +3685,6 @@ func TestFileStorePurgeExWithSubject(t *testing.T) {
 		require_Equal(t, state.Msgs, 2)
 		require_Equal(t, state.FirstSeq, 1)
 
-		// Capture the current index.db file if it exists.
-		sfile := filepath.Join(fcfg.StoreDir, msgDir, streamStreamStateFile)
-		buf, err := os.ReadFile(sfile)
-		require_NoError(t, err)
-		require_True(t, len(buf) > 0)
-
 		// Make sure we can recover same state.
 		fs.Stop()
 		fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)

server/memstore.go

@@ -656,6 +656,23 @@ func (ms *memStore) AllLastSeqs() ([]uint64, error) {
 	return seqs, nil
 }
 
+// Helper to determine if the filter(s) represent all the subjects.
+// Most clients send in subjects even if they match the stream's ingest subjects.
+// Lock should be held.
+func (ms *memStore) filterIsAll(filters []string) bool {
+	if len(filters) != len(ms.cfg.Subjects) {
+		return false
+	}
+	// Sort so we can compare.
+	slices.Sort(filters)
+	for i, subj := range filters {
+		if !subjectIsSubsetMatch(ms.cfg.Subjects[i], subj) {
+			return false
+		}
+	}
+	return true
+}
+
 // MultiLastSeqs will return a sorted list of sequences that match all subjects presented in filters.
 // We will not exceed the maxSeq, which if 0 becomes the store's last sequence.
 func (ms *memStore) MultiLastSeqs(filters []string, maxSeq uint64, maxAllowed int) ([]uint64, error) {
@@ -666,6 +683,11 @@ func (ms *memStore) MultiLastSeqs(filters []string, maxSeq uint64, maxAllowed in
 		return nil, nil
 	}
 
+	// See if we can short circuit if we think they are asking for all last sequences and have no maxSeq or maxAllowed set.
+	if maxSeq == 0 && maxAllowed <= 0 && ms.filterIsAll(filters) {
+		return ms.AllLastSeqs()
+	}
+
 	// Implied last sequence.
 	if maxSeq == 0 {
 		maxSeq = ms.state.LastSeq

server/memstore_test.go

@@ -921,7 +921,7 @@ func TestMemStoreSkipMsgs(t *testing.T) {
 func TestMemStoreMultiLastSeqs(t *testing.T) {
 	cfg := &StreamConfig{
 		Name:     "zzz",
-		Subjects: []string{"foo.*"},
+		Subjects: []string{"foo.*", "bar.*"},
 		Storage:  MemoryStorage,
 	}
 	ms, err := newMemStore(cfg)
@@ -1259,162 +1259,6 @@ func TestMemStoreSubjectDeleteMarkers(t *testing.T) {
 	require_Equal(t, bytesToString(getHeader(JSMessageTTL, im.hdr)), "1s")
 }
 
-func TestMemStoreSubjectDeleteMarkersOnPurge(t *testing.T) {
-	t.SkipNow()
-
-	ms, err := newMemStore(
-		&StreamConfig{
-			Name: "zzz", Subjects: []string{"test.*"}, Storage: MemoryStorage,
-			MaxAge: time.Second, AllowMsgTTL: true,
-			SubjectDeleteMarkerTTL: time.Second,
-		},
-	)
-	require_NoError(t, err)
-	defer ms.Stop()
-
-	for i := 0; i < 10; i++ {
-		_, _, err := ms.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
-		require_NoError(t, err)
-	}
-
-	_, err = ms.Purge()
-	require_NoError(t, err)
-
-	for i := uint64(0); i < 10; i++ {
-		sm, err := ms.LoadMsg(11+i, nil)
-		require_NoError(t, err)
-		require_Equal(t, sm.subj, fmt.Sprintf("test.%d", i))
-		require_Equal(t, bytesToString(getHeader(JSMarkerReason, sm.hdr)), JSMarkerReasonPurge)
-		require_Equal(t, bytesToString(getHeader(JSMessageTTL, sm.hdr)), "1s")
-	}
-}
-
-func TestMemStoreSubjectDeleteMarkersOnPurgeEx(t *testing.T) {
-	t.SkipNow()
-
-	ms, err := newMemStore(
-		&StreamConfig{
-			Name: "zzz", Subjects: []string{"test.*"}, Storage: MemoryStorage,
-			MaxAge: time.Second, AllowMsgTTL: true,
-			SubjectDeleteMarkerTTL: time.Second,
-		},
-	)
-	require_NoError(t, err)
-	defer ms.Stop()
-
-	for i := 0; i < 10; i++ {
-		_, _, err := ms.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
-		require_NoError(t, err)
-	}
-
-	_, err = ms.PurgeEx("test.*", 1, 0)
-	require_NoError(t, err)
-
-	for i := uint64(0); i < 10; i++ {
-		sm, err := ms.LoadMsg(11+i, nil)
-		require_NoError(t, err)
-		require_Equal(t, sm.subj, fmt.Sprintf("test.%d", i))
-		require_Equal(t, bytesToString(getHeader(JSMarkerReason, sm.hdr)), JSMarkerReasonPurge)
-		require_Equal(t, bytesToString(getHeader(JSMessageTTL, sm.hdr)), "1s")
-	}
-}
-
-func TestMemStoreSubjectDeleteMarkersOnPurgeExNoMarkers(t *testing.T) {
-	t.SkipNow()
-
-	ms, err := newMemStore(
-		&StreamConfig{
-			Name: "zzz", Subjects: []string{"test.*"}, Storage: MemoryStorage,
-			MaxAge: time.Second, AllowMsgTTL: true,
-			SubjectDeleteMarkerTTL: time.Second,
-		},
-	)
-	require_NoError(t, err)
-	defer ms.Stop()
-
-	for i := 0; i < 10; i++ {
-		_, _, err := ms.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
-		require_NoError(t, err)
-	}
-
-	_, err = ms.PurgeEx("test.*", 1, 0)
-	require_NoError(t, err)
-
-	for i := uint64(0); i < 10; i++ {
-		_, err := ms.LoadMsg(11+i, nil)
-		require_Error(t, err)
-	}
-}
-
-func TestMemStoreSubjectDeleteMarkersOnCompact(t *testing.T) {
-	t.SkipNow()
-
-	ms, err := newMemStore(
-		&StreamConfig{
-			Name: "zzz", Subjects: []string{"test.*"}, Storage: MemoryStorage,
-			MaxAge: time.Second, AllowMsgTTL: true,
-			SubjectDeleteMarkerTTL: time.Second,
-		},
-	)
-	require_NoError(t, err)
-	defer ms.Stop()
-
-	for i := 0; i < 10; i++ {
-		_, _, err := ms.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
-		require_NoError(t, err)
-	}
-
-	_, err = ms.Compact(6)
-	require_NoError(t, err)
-
-	for i := uint64(6); i <= 15; i++ {
-		sm, err := ms.LoadMsg(i, nil)
-		require_NoError(t, err)
-		if i <= 10 {
-			require_Equal(t, sm.subj, fmt.Sprintf("test.%d", i-1))
-			require_Equal(t, bytesToString(getHeader(JSMarkerReason, sm.hdr)), _EMPTY_)
-			require_Equal(t, bytesToString(getHeader(JSMessageTTL, sm.hdr)), _EMPTY_)
-		} else {
-			require_Equal(t, sm.subj, fmt.Sprintf("test.%d", 15-i))
-			require_Equal(t, bytesToString(getHeader(JSMarkerReason, sm.hdr)), JSMarkerReasonPurge)
-			require_Equal(t, bytesToString(getHeader(JSMessageTTL, sm.hdr)), "1s")
-		}
-	}
-}
-
-func TestMemStoreSubjectDeleteMarkersOnRemoveMsg(t *testing.T) {
-	t.SkipNow()
-
-	ms, err := newMemStore(
-		&StreamConfig{
-			Name: "zzz", Subjects: []string{"test"}, Storage: MemoryStorage,
-			MaxAge: time.Second, AllowMsgTTL: true,
-			SubjectDeleteMarkerTTL: time.Second,
-		},
-	)
-	require_NoError(t, err)
-	defer ms.Stop()
-
-	_, _, err = ms.StoreMsg("test", nil, nil, 0)
-	require_NoError(t, err)
-
-	_, err = ms.RemoveMsg(1)
-	require_NoError(t, err)
-
-	sm, err := ms.LoadMsg(2, nil)
-	require_NoError(t, err)
-	require_Equal(t, sm.subj, "test")
-	require_Equal(t, bytesToString(getHeader(JSMarkerReason, sm.hdr)), JSMarkerReasonRemove)
-	require_Equal(t, bytesToString(getHeader(JSMessageTTL, sm.hdr)), "1s")
-
-	_, err = ms.RemoveMsg(2)
-	require_NoError(t, err)
-
-	// The deleted subject marker at seq 2 should not have been replaced.
-	_, err = ms.LoadMsg(3, nil)
-	require_Error(t, err)
-}
-
 func TestMemStoreAllLastSeqs(t *testing.T) {
 	cfg := &StreamConfig{
 		Name:       "zzz",