Repository: nats-io/nats-server
Pull request #5381 (2024-05-02 16:58:12)

Cherry picks for v2.10.15 RC.6
Includes:

- https://github.com/nats-io/nats-server/pull/5378
- https://github.com/nats-io/nats-server/pull/5379
- https://github.com/nats-io/nats-server/pull/5380
- https://github.com/nats-io/nats-server/pull/5374



Patch:
diff --git a/server/events.go b/server/events.go
index ed3330f4..3e257eab 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1120,6 +1120,10 @@ func (s *Server) initEventTracking() {
 			optz := &ProfilezEventOptions{}
 			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.profilez(&optz.ProfilezOptions), nil })
 		},
+		"EXPVARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+			optz := &ExpvarzEventOptions{}
+			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.expvarz(optz), nil })
+		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
@@ -1788,6 +1792,11 @@ type ProfilezEventOptions struct {
 	EventFilterOptions
 }
 
+// In the context of system events, ExpvarzEventOptions are options passed to Expvarz
+type ExpvarzEventOptions struct {
+	EventFilterOptions
+}
+
 // returns true if the request does NOT apply to this server and can be ignored.
 // DO NOT hold the server lock when
 func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
diff --git a/server/filestore.go b/server/filestore.go
index 3ba9cc3f..cd12da19 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -219,6 +219,7 @@ type msgBlock struct {
 	lwts       int64
 	llts       int64
 	lrts       int64
+	lsts       int64
 	llseq      uint64
 	hh         hash.Hash64
 	cache      *cache
@@ -2309,6 +2310,8 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		mb.loadMsgsWithLock()
 		didLoad = true
 	}
+	// Mark fss activity.
+	mb.lsts = time.Now().UnixNano()
 
 	// If we only have 1 subject currently and it matches our filter we can also set isAll.
 	if !isAll && len(mb.fss) == 1 {
@@ -2667,6 +2670,8 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 			mb.loadMsgsWithLock()
 			shouldExpire = true
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				if ss.firstNeedsUpdate {
@@ -2846,6 +2851,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 				mb.loadMsgsWithLock()
 				shouldExpire = true
 			}
+			// Mark fss activity.
+			mb.lsts = time.Now().UnixNano()
+
 			var havePartial bool
 			for subj, ss := range mb.fss {
 				if isMatch(subj) {
@@ -2861,6 +2869,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					}
 				}
 			}
+
 			// See if we need to scan msgs here.
 			if havePartial {
 				// Make sure we have the cache loaded.
@@ -2934,6 +2943,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					mb.loadMsgsWithLock()
 					shouldExpire = true
 				}
+				// Mark fss activity.
+				mb.lsts = time.Now().UnixNano()
+
 				for subj, ss := range mb.fss {
 					if isMatch(subj) {
 						adjust += ss.Msgs
@@ -3501,6 +3513,9 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 			}
 			shouldExpire = true
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
+
 		if ss := mb.fss[subj]; ss != nil {
 			// Adjust first if it was not where we thought it should be.
 			if i != start {
@@ -4727,6 +4742,8 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
 			return err
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		if ss := mb.fss[subj]; ss != nil {
 			ss.Msgs++
 			ss.Last = seq
@@ -5335,6 +5352,8 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		mb.fss = make(map[string]*SimpleState)
 		popFss = true
 	}
+	// Mark fss activity.
+	mb.lsts = time.Now().UnixNano()
 
 	lbuf := uint32(len(buf))
 	var seq uint64
@@ -6099,6 +6118,9 @@ func (fs *fileStore) loadLast(subj string, sm *StoreMsg) (lsm *StoreMsg, err err
 			mb.mu.Unlock()
 			return nil, err
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
+
 		var l uint64
 		// Optimize if subject is not a wildcard.
 		if !wc {
@@ -6347,6 +6369,9 @@ func (mb *msgBlock) sinceLastActivity() time.Duration {
 	if mb.llts > last {
 		last = mb.llts
 	}
+	if mb.lsts > last {
+		last = mb.lsts
+	}
 	return time.Since(time.Unix(0, last).UTC())
 }
 
@@ -7329,6 +7354,8 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 	if len(mb.fss) > 0 {
 		// Make sure we run the cache expire timer.
 		mb.llts = time.Now().UnixNano()
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		mb.startCacheExpireTimer()
 	}
 	return nil
@@ -7338,6 +7365,10 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 // Lock should be held
 func (mb *msgBlock) ensurePerSubjectInfoLoaded() error {
 	if mb.fss != nil || mb.noTrack {
+		if mb.fss != nil {
+			// Mark fss activity.
+			mb.lsts = time.Now().UnixNano()
+		}
 		return nil
 	}
 	if mb.msgs == 0 {
@@ -7712,36 +7743,24 @@ func (fs *fileStore) _writeFullState(force bool) error {
 		fs.debug("WriteFullState reallocated from %d to %d", sz, cap(buf))
 	}
 
-	// Write to a tmp file and rename.
-	const tmpPre = streamStreamStateFile + tsep
-	f, err := os.CreateTemp(filepath.Join(fs.fcfg.StoreDir, msgDir), tmpPre)
-	if err != nil {
-		return err
-	}
-	tmpName := f.Name()
-	defer os.Remove(tmpName)
-	if _, err = f.Write(buf); err == nil && fs.fcfg.SyncAlways {
-		f.Sync()
-	}
-	f.Close()
-	if err != nil {
-		return err
+	// Only warn about construction time since file write not holding any locks.
+	if took := time.Since(start); took > time.Minute {
+		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
 	}
 
-	// Rename into position under our lock, clear prior dirty pending on success.
-	fs.mu.Lock()
-	if !fs.closed {
-		if err := os.Rename(tmpName, fn); err != nil {
-			fs.mu.Unlock()
-			return err
-		}
+	// Write our update index.db
+	// Protect with dios.
+	<-dios
+	err := os.WriteFile(fn, buf, defaultFilePerms)
+	dios <- struct{}{}
+
+	// Update dirty if successful.
+	if err == nil {
+		fs.mu.Lock()
 		fs.dirty -= priorDirty
+		fs.mu.Unlock()
 	}
-	fs.mu.Unlock()
 
-	if took := time.Since(start); took > time.Minute {
-		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
-	}
 	return nil
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index 3da0930d..0467d821 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -20,6 +20,7 @@ import (
 	"crypto/x509"
 	"encoding/hex"
 	"encoding/json"
+	"expvar"
 	"fmt"
 	"net"
 	"net/http"
@@ -2202,9 +2203,9 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 	}
 	s.mu.Unlock()
 
-	var leafnodes []*LeafInfo
+	leafnodes := make([]*LeafInfo, 0, len(lconns))
+
 	if len(lconns) > 0 {
-		leafnodes = make([]*LeafInfo, 0, len(lconns))
 		for _, ln := range lconns {
 			ln.mu.Lock()
 			lni := &LeafInfo{
@@ -2231,6 +2232,7 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 			leafnodes = append(leafnodes, lni)
 		}
 	}
+
 	return &Leafz{
 		ID:       s.ID(),
 		Now:      time.Now().UTC(),
@@ -3671,6 +3673,30 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	return health
 }
 
+type ExpvarzStatus struct {
+	Memstats json.RawMessage `json:"memstats"`
+	Cmdline  json.RawMessage `json:"cmdline"`
+}
+
+func (s *Server) expvarz(_ *ExpvarzEventOptions) *ExpvarzStatus {
+	var stat ExpvarzStatus
+
+	const memStatsKey = "memstats"
+	const cmdLineKey = "cmdline"
+
+	expvar.Do(func(v expvar.KeyValue) {
+		switch v.Key {
+		case memStatsKey:
+			stat.Memstats = json.RawMessage(v.Value.String())
+
+		case cmdLineKey:
+			stat.Cmdline = json.RawMessage(v.Value.String())
+		}
+	})
+
+	return &stat
+}
+
 type ProfilezStatus struct {
 	Profile []byte `json:"profile"`
 	Error   string `json:"error"`
diff --git a/server/stream.go b/server/stream.go
index 50500d2a..20f441a1 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2463,6 +2463,9 @@ func (mset *stream) scheduleSetupMirrorConsumerRetry() {
 	})
 }
 
+// How long we wait for a response from a consumer create request for a source or mirror.
+var srcConsumerWaitTime = 30 * time.Second
+
 // Setup our mirror consumer.
 // Lock should be held.
 func (mset *stream) setupMirrorConsumer() error {
@@ -2721,9 +2724,9 @@ func (mset *stream) setupMirrorConsumer() error {
 			}
 			mset.mu.Unlock()
 			ready.Wait()
-		case <-time.After(5 * time.Second):
+		case <-time.After(srcConsumerWaitTime):
 			mset.unsubscribe(crSub)
-			// We already waited 5 seconds, let's retry now.
+			// We already waited 30 seconds, let's retry now.
 			retry = true
 		}
 	}()
@@ -3078,9 +3081,9 @@ func (mset *stream) setupSourceConsumer(iname string, seq uint64, startTime time
 			}
 			mset.mu.Unlock()
 			ready.Wait()
-		case <-time.After(5 * time.Second):
+		case <-time.After(srcConsumerWaitTime):
 			mset.unsubscribe(crSub)
-			// We already waited 5 seconds, let's retry now.
+			// We already waited 30 seconds, let's retry now.
 			retry = true
 		}
 	}()
@@ -3298,7 +3301,9 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 			if errors.Is(err, ErrMaxMsgs) {
 				// Do not need to do a full retry that includes finding the last sequence in the stream
 				// for that source. Just re-create starting with the seq we couldn't store instead.
+				mset.mu.Lock()
 				mset.retrySourceConsumerAtSeq(iname, si.sseq)
+				mset.mu.Unlock()
 			} else {
 				// Log some warning for errors other than errLastSeqMismatch or errMaxMsgs.
 				if !errors.Is(err, errLastSeqMismatch) {


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 400b9d03..7e6e7b9e 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1674,8 +1674,8 @@ func TestSystemAccountWithGateways(t *testing.T) {
 	nca.Flush()
 
 	// If this tests fails with wrong number after 10 seconds we may have
-	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 56, sa)
+	// added a new initial subscription for the eventing system.
+	checkExpectedSubs(t, 58, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
@@ -2990,6 +2990,7 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 		{"HEALTHZ", nil, &JSzOptions{}, []string{"status"}},
 		{"HEALTHZ", &HealthzOptions{JSEnabledOnly: true}, &JSzOptions{}, []string{"status"}},
 		{"HEALTHZ", &HealthzOptions{JSServerOnly: true}, &JSzOptions{}, []string{"status"}},
+		{"EXPVARZ", nil, &ExpvarzStatus{}, []string{"memstats", "cmdline"}},
 	}
 
 	for i, test := range tests {
@@ -3058,8 +3059,10 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 					strings.ToLower(test.endpoint), string(msg.Data))
 			} else {
 				for _, respField := range test.respField {
-					if _, ok := resp[respField]; !ok {
+					if val, ok := resp[respField]; !ok {
 						t.Fatalf("Error finding: %s in %s", respField, resp)
+					} else if val == nil {
+						t.Fatalf("Nil value found: %s in %s", respField, resp)
 					}
 				}
 			}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 95e0c5b2..95af7639 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6687,9 +6687,9 @@ func TestFileStoreWriteFullStateAfterPurgeEx(t *testing.T) {
 	require_Equal(t, ss.LastSeq, 10)
 }
 
-func TestFileStoreMB_FSS_Expire(t *testing.T) {
+func TestFileStoreFSSExpire(t *testing.T) {
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SyncInterval: 2 * time.Second},
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SubjectStateExpire: time.Second},
 		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, MaxMsgsPer: 1, Storage: FileStorage})
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -6724,6 +6724,78 @@ func TestFileStoreMB_FSS_Expire(t *testing.T) {
 	require_True(t, cache != nil)
 }
 
+func TestFileStoreFSSExpireNumPending(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 8192, CacheExpire: 1 * time.Second, SubjectStateExpire: 2 * time.Second},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, MaxMsgsPer: 1, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 100_000; i++ {
+		fs.StoreMsg(fmt.Sprintf("foo.A.%d", i), nil, msg)
+		fs.StoreMsg(fmt.Sprintf("foo.B.%d", i), nil, msg)
+	}
+	// Flush fss by hand, cache should be flushed as well.
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		mb.fss = nil
+		mb.mu.Unlock()
+	}
+	fs.mu.RUnlock()
+
+	nb := fs.numMsgBlocks()
+	// Now execute NumPending() such that we load lots of blocks and make sure fss do not expire.
+	start := time.Now()
+	n, _ := fs.NumPending(100_000, "foo.A.*", false)
+	elapsed := time.Since(start)
+
+	require_Equal(t, n, 50_000)
+	// Make sure we did not force expire the fss. We would have loaded first half of blocks.
+	var noFss bool
+	last := nb/2 - 1
+	fs.mu.RLock()
+	for i, mb := range fs.blks {
+		mb.mu.RLock()
+		noFss = mb.fss == nil
+		mb.mu.RUnlock()
+		if noFss || i == last {
+			break
+		}
+	}
+	fs.mu.RUnlock()
+	require_False(t, noFss)
+
+	// Run again, make sure faster. This is consequence of fss being loaded now.
+	start = time.Now()
+	fs.NumPending(100_000, "foo.A.*", false)
+	require_True(t, elapsed > 2*time.Since(start))
+
+	// Now do with start past the mid-point.
+	start = time.Now()
+	fs.NumPending(150_000, "foo.B.*", false)
+	elapsed = time.Since(start)
+	time.Sleep(time.Second)
+	start = time.Now()
+	fs.NumPending(150_000, "foo.B.*", false)
+	require_True(t, elapsed > time.Since(start))
+
+	// Sleep enough so that all mb.fss should expire, which is 2s above.
+	time.Sleep(3 * time.Second)
+	fs.mu.RLock()
+	for i, mb := range fs.blks {
+		mb.mu.RLock()
+		fss := mb.fss
+		mb.mu.RUnlock()
+		if fss != nil {
+			fs.mu.RUnlock()
+			t.Fatalf("Detected loaded fss for mb %d", i)
+		}
+	}
+	fs.mu.RUnlock()
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 31353741..578e54a8 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5373,6 +5373,10 @@ func TestJetStreamClusterLeaderStepdown(t *testing.T) {
 }
 
 func TestJetStreamClusterSourcesFilteringAndUpdating(t *testing.T) {
+	owt := srcConsumerWaitTime
+	srcConsumerWaitTime = 2 * time.Second
+	defer func() { srcConsumerWaitTime = owt }()
+
 	c := createJetStreamClusterExplicit(t, "MSR", 5)
 	defer c.shutdown()
 
@@ -5589,6 +5593,10 @@ func TestJetStreamClusterSourcesUpdateOriginError(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
+	owt := srcConsumerWaitTime
+	srcConsumerWaitTime = 2 * time.Second
+	defer func() { srcConsumerWaitTime = owt }()
+
 	test := func(t *testing.T, mirror bool, filter bool) {
 		c := createJetStreamClusterExplicit(t, "MSR", 5)
 		defer c.shutdown()
@@ -5643,7 +5651,7 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 
 		checkSync := func(msgsTest, msgsM uint64) {
 			t.Helper()
-			checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
+			checkFor(t, 40*time.Second, time.Second, func() error {
 				if tsi, err := js.StreamInfo("TEST"); err != nil {
 					return err
 				} else if msi, err := js.StreamInfo("M"); err != nil {
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index a14ed5d9..96b03765 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -5317,6 +5317,10 @@ func TestJetStreamClusterMirrorSourceLoop(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorDeDupWindow(t *testing.T) {
+	owt := srcConsumerWaitTime
+	srcConsumerWaitTime = 2 * time.Second
+	defer func() { srcConsumerWaitTime = owt }()
+
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 19514407..79c82ada 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -4044,7 +4044,7 @@ func TestMonitorAccountz(t *testing.T) {
 	body = string(readBody(t, fmt.Sprintf("http://127.0.0.1:%d%s?acc=$SYS", s.MonitorAddr().Port, AccountzPath)))
 	require_Contains(t, body, `"account_detail": {`)
 	require_Contains(t, body, `"account_name": "$SYS",`)
-	require_Contains(t, body, `"subscriptions": 50,`)
+	require_Contains(t, body, `"subscriptions": 52,`)
 	require_Contains(t, body, `"is_system": true,`)
 	require_Contains(t, body, `"system_account": "$SYS"`)
 
diff --git a/server/norace_test.go b/server/norace_test.go
index 14ae42a1..9d61625e 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1767,6 +1767,10 @@ func TestNoRaceJetStreamSuperClusterMixedModeMirrors(t *testing.T) {
 }
 
 func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
+	owt := srcConsumerWaitTime
+	srcConsumerWaitTime = 2 * time.Second
+	defer func() { srcConsumerWaitTime = owt }()
+
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
 
@@ -9266,6 +9270,10 @@ func TestNoRaceJetStreamAPIDispatchQueuePending(t *testing.T) {
 }
 
 func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
+	owt := srcConsumerWaitTime
+	srcConsumerWaitTime = 2 * time.Second
+	defer func() { srcConsumerWaitTime = owt }()
+
 	// Check calculations first.
 	for i := 1; i <= 20; i++ {
 		backoff := calculateRetryBackoff(i)
@@ -9322,8 +9330,8 @@ func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
 	sldr := c.streamLeader(globalAccountName, "TEST")
 	sldr.Shutdown()
 
-	// Wait for just greater than 10s. We should only see 1 request during this time.
-	time.Sleep(11 * time.Second)
+	// Wait for just greater than 5s. We should only see 1 request during this time.
+	time.Sleep(6 * time.Second)
 	// There should have been 2 requests, one for mirror, one for source
 	n, _, _ := sub.Pending()
 	require_Equal(t, n, 2)