nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: nats-io/nats-server
Pull request #4346 (2023-07-28 20:27:19)

[FIXED] Do not panic when re-calculating first sequence for SimpleState when fseq moves ahead of old first.
When a lazy simple state has an outdated first that needs to be updated, if fseq had moved past it would panic.
This was not common but with latest fix prior in can become more common, hence why it showed up.

Signed-off-by: Derek Collison <derek@nats.io>




Patch:
diff --git a/server/filestore.go b/server/filestore.go
index a8607a08..c95b1126 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -5934,6 +5934,8 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si
 	if startSlot >= len(mb.cache.idx) {
 		ss.First = ss.Last
 		return
+	} else if startSlot < 0 {
+		startSlot = 0
 	}
 
 	var le = binary.LittleEndian


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 973933bd..6c902f2a 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -5537,3 +5537,35 @@ func TestFileStoreRestoreEncryptedWithNoKeyFuncFails(t *testing.T) {
 	)
 	require_Error(t, err, errNoMainKey)
 }
+
+func TestFileStoreRecaluclateFirstForSubjBug(t *testing.T) {
+	fs, err := newFileStore(FileStoreConfig{StoreDir: t.TempDir()}, StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
+	require_NoError(t, err)
+
+	fs.StoreMsg("foo", nil, nil) // 1
+	fs.StoreMsg("bar", nil, nil) // 2
+	fs.StoreMsg("foo", nil, nil) // 3
+
+	// Now remove first 2..
+	fs.RemoveMsg(1)
+	fs.RemoveMsg(2)
+
+	// Now grab first (and only) block.
+	fs.mu.RLock()
+	mb := fs.blks[0]
+	fs.mu.RUnlock()
+
+	// Since we lazy update the first, simulate that we have not updated it as of yet.
+	ss := &SimpleState{Msgs: 1, First: 1, Last: 3, firstNeedsUpdate: true}
+
+	mb.mu.Lock()
+	defer mb.mu.Unlock()
+
+	// Flush the cache.
+	mb.clearCacheAndOffset()
+	// Now call with start sequence of 1, the old one
+	// This will panic without the fix.
+	mb.recalculateFirstForSubj("foo", 1, ss)
+	// Make sure it was update properly.
+	require_True(t, *ss == SimpleState{Msgs: 1, First: 3, Last: 3, firstNeedsUpdate: false})
+}
Repository: nats-io/nats-server
Pull request #4439 (2023-08-27 20:48:38)

[FIXED] Expire on recover could not update global per subject map
When expiring complete blocks on recover make sure to update global subject index psim.

Signed-off-by: Derek Collison <derek@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index e30e8a77..36594ccc 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1315,6 +1315,11 @@ func (fs *fileStore) expireMsgsOnRecover() {
 			fs.psim = make(map[string]*psi)
 			return false
 		}
+		// Make sure we do subject cleanup as well.
+		mb.ensurePerSubjectInfoLoaded()
+		for subj := range mb.fss {
+			fs.removePerSubject(subj)
+		}
 		mb.dirtyCloseWithRemove(true)
 		deleted++
 		return true


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 81032303..a30f9942 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4662,6 +4662,40 @@ func TestFileStoreFSSCloseAndKeepOnExpireOnRecoverBug(t *testing.T) {
 	})
 }
 
+func TestFileStoreExpireOnRecoverSubjectAccounting(t *testing.T) {
+	const msgLen = 19
+	msg := bytes.Repeat([]byte("A"), msgLen)
+
+	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+		fcfg.BlockSize = 100
+		ttl := 200 * time.Millisecond
+		scfg := StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage, MaxAge: ttl}
+
+		fs, err := newFileStore(fcfg, scfg)
+		require_NoError(t, err)
+		defer fs.Stop()
+
+		// These are in first block.
+		fs.StoreMsg("A", nil, msg)
+		fs.StoreMsg("B", nil, msg)
+		time.Sleep(ttl / 2)
+		// This one in 2nd block.
+		fs.StoreMsg("C", nil, msg)
+		fs.Stop()
+
+		time.Sleep(ttl/2 + 10*time.Millisecond)
+
+		fs, err = newFileStore(fcfg, scfg)
+		require_NoError(t, err)
+		defer fs.Stop()
+
+		// Make sure we take into account PSIM when throwing a whole block away.
+		if state := fs.State(); state.NumSubjects != 1 {
+			t.Fatalf("Expected 1 subject, got %d", state.NumSubjects)
+		}
+	})
+}
+
 func TestFileStoreFSSBadStateBug(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		fs, err := newFileStore(
Repository: nats-io/nats-server
Pull request #5485 (2024-06-05 02:03:43)

Fixes for tests that depend on fast statsz
Also minor fix for new test to cleanup cluster when done.

Signed-off-by: Derek Collison <derek@nats.io>




Patch:
diff --git a/server/events.go b/server/events.go
index 24bc0f06..a2f6c570 100644
--- a/server/events.go
+++ b/server/events.go
@@ -98,7 +98,11 @@ const (
 // FIXME(dlc) - make configurable.
 var eventsHBInterval = 30 * time.Second
 
-var statszRateLimit = 1 * time.Second
+// Default minimum wait time for sending statsz
+const defaultStatszRateLimit = 1 * time.Second
+
+// Variable version so we can set in tests.
+var statszRateLimit = defaultStatszRateLimit
 
 type sysMsgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, hdr, msg []byte)
 


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 8aed50fe..0d52ca20 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -3540,10 +3540,16 @@ func Benchmark_GetHash(b *testing.B) {
 }
 
 func TestClusterSetupMsgs(t *testing.T) {
+	// Tests will set this general faster, but here we want default for production.
+	original := statszRateLimit
+	statszRateLimit = defaultStatszRateLimit
+	defer func() { statszRateLimit = original }()
+
 	numServers := 10
 	c := createClusterEx(t, false, 0, false, "cluster", numServers)
-	var totalOut int
+	defer shutdownCluster(c)
 
+	var totalOut int
 	for _, server := range c.servers {
 		totalOut += int(server.outMsgs)
 	}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index f130b524..3d63b0b5 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -42,6 +42,9 @@ func init() {
 	maxElectionTimeout = 3500 * time.Millisecond
 	lostQuorumInterval = 2 * time.Second
 	lostQuorumCheck = 4 * hbInterval
+
+	// For statz and jetstream placement speedups as well.
+	statszRateLimit = time.Millisecond * 100
 }
 
 // Used to setup clusters of clusters for tests.
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index 9c2c1fd1..ffd7c122 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -1355,10 +1355,6 @@ func TestJetStreamSuperClusterPushConsumerInterest(t *testing.T) {
 }
 
 func TestJetStreamSuperClusterOverflowPlacement(t *testing.T) {
-	orgStatszRateLimit := statszRateLimit
-	statszRateLimit = time.Millisecond * 100
-	defer func() { statszRateLimit = orgStatszRateLimit }()
-
 	sc := createJetStreamSuperClusterWithTemplate(t, jsClusterMaxBytesTempl, 3, 3)
 	defer sc.shutdown()