Repository: nats-io/nats-server
Pull request #3416 (2022-08-30 12:33:10)

When writing a msg after the fss state was expired we would count the msg twice.
This was due to the fact that we updated the accounting and then checked if we needed to load state which would then fallback to generating the state on the fly.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 727400d6..925ed368 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1848,11 +1848,10 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	// Race detector wants these protected.
 	mb.mu.Lock()
 	mb.llts, mb.lwts = ts, ts
-	mb.mu.Unlock()
-
 	// Remember our last sequence number.
 	mb.first.seq = fs.state.LastSeq + 1
 	mb.last.seq = fs.state.LastSeq
+	mb.mu.Unlock()
 
 	// If we know we will need this so go ahead and spin up.
 	if !fs.fip {
@@ -3183,22 +3182,20 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 	// Decide if we write index info if flushing in place.
 	writeIndex := ts-mb.lwits > wiThresh
 
-	// Accounting
-	mb.updateAccounting(seq, ts, rl)
-
 	// Check if we are tracking per subject for our simple state.
-	if len(subj) > 0 {
+	if len(subj) > 0 && !mb.noTrack {
 		mb.ensurePerSubjectInfoLoaded()
-		if mb.fss != nil {
-			if ss := mb.fss[subj]; ss != nil {
-				ss.Msgs++
-				ss.Last = seq
-			} else {
-				mb.fss[subj] = &SimpleState{Msgs: 1, First: seq, Last: seq}
-			}
+		if ss := mb.fss[subj]; ss != nil {
+			ss.Msgs++
+			ss.Last = seq
+		} else {
+			mb.fss[subj] = &SimpleState{Msgs: 1, First: seq, Last: seq}
 		}
 	}
 
+	// Accounting
+	mb.updateAccounting(seq, ts, rl)
+
 	fch, werr := mb.fch, mb.werr
 
 	// If we should be flushing, or had a write error, do so here.
@@ -5197,10 +5194,15 @@ func (mb *msgBlock) loadPerSubjectInfo() ([]byte, error) {
 // Helper to make sure fss loaded if we are tracking.
 // Lock should be held
 func (mb *msgBlock) ensurePerSubjectInfoLoaded() error {
-	if mb.fss == nil && !mb.noTrack {
-		return mb.readPerSubjectInfo(true)
+	if mb.fss != nil || mb.noTrack {
+		return nil
 	}
-	return nil
+	if mb.msgs == 0 {
+		mb.fss = make(map[string]*SimpleState)
+		return nil
+	}
+	// Load from file.
+	return mb.readPerSubjectInfo(true)
 }
 
 // Called on recovery to populate the global psim state.


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 23b25da5..846b7b63 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3506,7 +3506,9 @@ func TestFileStorePurgeExKeepOneBug(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
-	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 128}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 128},
+		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4586,3 +4588,26 @@ func TestFileStoreFSSBadStateBug(t *testing.T) {
 		t.Fatalf("Got bad state on restart: %+v", fss)
 	}
 }
+
+func TestFileStoreFSSExpireNumPendingBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	cexp := 100 * time.Millisecond
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp},
+		StreamConfig{Name: "zzz", Subjects: []string{"KV.>"}, MaxMsgsPer: 1, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Let FSS meta expire.
+	time.Sleep(2 * cexp)
+
+	_, _, err = fs.StoreMsg("KV.X", nil, []byte("Y"))
+	require_NoError(t, err)
+
+	if fss := fs.FilteredState(1, "KV.X"); fss.Msgs != 1 {
+		t.Fatalf("Expected only 1 msg, got %d", fss.Msgs)
+	}
+}
diff --git a/server/norace_test.go b/server/norace_test.go
index ff730b4d..1b3d7a03 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -5399,7 +5399,7 @@ func TestNoRaceJetStreamFileStoreLargeKVAccessTiming(t *testing.T) {
 
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize, CacheExpire: 5 * time.Second},
-		StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgsPer: 1},
+		StreamConfig{Name: "zzz", Subjects: []string{"KV.STREAM_NAME.*"}, Storage: FileStorage, MaxMsgsPer: 1},
 	)
 	require_NoError(t, err)
 	defer fs.Stop()