Repository: nats-io/nats-server
Pull request #3366 (2022-08-12 03:49:25)

[FIXED] Make sure when SubjectState is called we have loaded fss state.
Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 21ee18f1..5a745415 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1536,9 +1536,29 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 		return nil
 	}
 
+	start, stop := fs.blks[0], fs.lmb
+	// We can short circuit if not a wildcard using psim for start and stop.
+	if !subjectHasWildcard(subject) {
+		info := fs.psim[subject]
+		if info == nil {
+			return nil
+		}
+		start, stop = fs.bim[info.fblk], fs.bim[info.lblk]
+	}
+
 	fss := make(map[string]SimpleState)
+	var startFound bool
+
 	for _, mb := range fs.blks {
-		mb.mu.RLock()
+		if !startFound {
+			if mb != start {
+				continue
+			}
+			startFound = true
+		}
+
+		mb.mu.Lock()
+		mb.ensurePerSubjectInfoLoaded()
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				oss := fss[subj]
@@ -1551,8 +1571,13 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 				}
 			}
 		}
-		mb.mu.RUnlock()
+		mb.mu.Unlock()
+
+		if mb == stop {
+			break
+		}
 	}
+
 	return fss
 }
 
@@ -1726,9 +1751,6 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts in
 	var psmc uint64
 	psmax := fs.cfg.MaxMsgsPer > 0 && len(subj) > 0
 	if psmax {
-
-		// FIXME(dlc) = psim nil??
-
 		if info, ok := fs.psim[subj]; ok {
 			psmc = info.total
 		}
diff --git a/server/stream.go b/server/stream.go
index e85c68bd..aa903907 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -402,7 +402,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	// Check for overlapping subjects. These are not allowed for now.
 	if jsa.subjectsOverlap(cfg.Subjects) {
 		jsa.mu.Unlock()
-		return nil, NewJSStreamSubjectOverlapError()
+		return nil, fmt.Errorf("subjects overlap with an existing stream")
 	}
 
 	if !hasTier {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 1aa99599..b3d0e7db 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4241,3 +4241,65 @@ func TestFileStoreMaxMsgsPerSubject(t *testing.T) {
 		t.Fatalf("Expected 34 blocks, got %d", nb)
 	}
 }
+
+func TestFileStoreSubjectStateCacheExpiration(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 32, CacheExpire: time.Second},
+		StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 2},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 1; i <= 100; i++ {
+		subj := fmt.Sprintf("kv.foo.%d", i)
+		_, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+		require_NoError(t, err)
+	}
+	for i := 1; i <= 100; i++ {
+		subj := fmt.Sprintf("kv.bar.%d", i)
+		_, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+		require_NoError(t, err)
+	}
+
+	// Make sure we clear mb fss meta before asking for SubjectState.
+	checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+		if _, hasAnyFSS := fs.reportMeta(); hasAnyFSS {
+			return fmt.Errorf("Still have mb fss state")
+		}
+		return nil
+	})
+
+	if fss := fs.SubjectsState("kv.bar.>"); len(fss) != 100 {
+		t.Fatalf("Expected 100 entries but got %d", len(fss))
+	}
+
+	fss := fs.SubjectsState("kv.bar.99")
+	if len(fss) != 1 {
+		t.Fatalf("Expected 1 entry but got %d", len(fss))
+	}
+	expected := SimpleState{Msgs: 1, First: 199, Last: 199}
+	if ss := fss["kv.bar.99"]; ss != expected {
+		t.Fatalf("Bad subject state, expected %+v but got %+v", expected, ss)
+	}
+
+	// Now add one to end and check as well for non-wildcard.
+	_, _, err = fs.StoreMsg("kv.foo.1", nil, []byte("value22"))
+	require_NoError(t, err)
+
+	if state := fs.State(); state.Msgs != 201 {
+		t.Fatalf("Expected 201 msgs but got %+v", state)
+	}
+
+	fss = fs.SubjectsState("kv.foo.1")
+	if len(fss) != 1 {
+		t.Fatalf("Expected 1 entry but got %d", len(fss))
+	}
+	expected = SimpleState{Msgs: 2, First: 1, Last: 201}
+	if ss := fss["kv.foo.1"]; ss != expected {
+		t.Fatalf("Bad subject state, expected %+v but got %+v", expected, ss)
+	}
+
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 0e6ce28f..6c080c63 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -18908,45 +18908,3 @@ func TestJetStreamDirectGetAutoSet(t *testing.T) {
 		t.Fatalf("Expected to see the mirror respond at least once")
 	}
 }
-
-func TestJetStreamProperErrorDueToOverlapSubjects(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-	defer s.Shutdown()
-
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	test := func(t *testing.T, s *Server) {
-		nc, js := jsClientConnect(t, s)
-		defer nc.Close()
-
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     "TEST",
-			Subjects: []string{"foo.*"},
-		})
-		require_NoError(t, err)
-
-		// Now do this by end since we want to check the error returned.
-		sc := &nats.StreamConfig{
-			Name:     "TEST2",
-			Subjects: []string{"foo.>"},
-		}
-		req, _ := json.Marshal(sc)
-		msg, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, sc.Name), req, time.Second)
-		require_NoError(t, err)
-
-		var scResp JSApiStreamCreateResponse
-		err = json.Unmarshal(msg.Data, &scResp)
-		require_NoError(t, err)
-
-		if scResp.Error == nil || !IsNatsErr(scResp.Error, JSStreamSubjectOverlapErr) {
-			t.Fatalf("Did not receive correct error: %+v", scResp)
-		}
-	}
-
-	t.Run("standalone", func(t *testing.T) { test(t, s) })
-	t.Run("clustered", func(t *testing.T) { test(t, c.randomServer()) })
-}