Repository: nats-io/nats-server
Pull request #3035 (2022-04-14 02:50:06)

[IMPROVED] JetStream: reduce unnecessary leader election
- Wait of some sort of routing to be in place before starting
the raft run loop
- Remove use of lock in apiDispatch that was not necessary but
could have cause a route to block, causing memory growth, etc..

Unrelated rename of some tests so that they start with TestJetStream
and TestJetStreamCluster for cluster tests, fixed some flappers
and ensure that tests that change RAFT timeouts put them back
to default values on exit.


Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index b5f8aa61..658b2875 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -651,9 +651,8 @@ const defaultMaxJSApiOut = int64(4096)
 var maxJSApiOut = defaultMaxJSApiOut
 
 func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, subject, reply string, rmsg []byte) {
-	js.mu.RLock()
+	// No lock needed, those are immutable.
 	s, rr := js.srv, js.apiSubs.Match(subject)
-	js.mu.RUnlock()
 
 	hdr, _ := c.msgParts(rmsg)
 	if len(getHeader(ClientInfoHdr, hdr)) == 0 {
diff --git a/server/raft.go b/server/raft.go
index c8672240..ab297bd8 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -236,13 +236,28 @@ const (
 )
 
 var (
+	minElectionTimeout time.Duration
+	maxElectionTimeout time.Duration
+	minCampaignTimeout time.Duration
+	maxCampaignTimeout time.Duration
+	hbInterval         time.Duration
+	lostQuorumInterval time.Duration
+)
+
+func init() {
+	setDefaultRaftTimeouts()
+}
+
+// This is called on init, but also by tests that want to tweak the settings
+// and then call this as a defer to restore the defaults.
+func setDefaultRaftTimeouts() {
 	minElectionTimeout = minElectionTimeoutDefault
 	maxElectionTimeout = maxElectionTimeoutDefault
 	minCampaignTimeout = minCampaignTimeoutDefault
 	maxCampaignTimeout = maxCampaignTimeoutDefault
-	hbInterval         = hbIntervalDefault
+	hbInterval = hbIntervalDefault
 	lostQuorumInterval = lostQuorumIntervalDefault
-)
+}
 
 type RaftConfig struct {
 	Name     string
@@ -1528,6 +1543,10 @@ func (n *raft) resetElectionTimeout() {
 	n.resetElect(randElectionTimeout())
 }
 
+func (n *raft) resetElectionTimeoutWithLock() {
+	n.resetElectWithLock(randElectionTimeout())
+}
+
 // Lock should be held.
 func (n *raft) resetElect(et time.Duration) {
 	if n.elect == nil {
@@ -1543,10 +1562,41 @@ func (n *raft) resetElect(et time.Duration) {
 	}
 }
 
+func (n *raft) resetElectWithLock(et time.Duration) {
+	n.Lock()
+	n.resetElect(et)
+	n.Unlock()
+}
+
 func (n *raft) run() {
 	s := n.s
 	defer s.grWG.Done()
 
+	// We want to wait for some routing to be enabled, so we will wait for
+	// at least a route, leaf or gateway connection to be established before
+	// starting the run loop.
+	gw := s.gateway
+	for {
+		s.mu.Lock()
+		ready := len(s.routes)+len(s.leafs) > 0
+		if !ready && gw.enabled {
+			gw.RLock()
+			ready = len(gw.out)+len(gw.in) > 0
+			gw.RUnlock()
+		}
+		s.mu.Unlock()
+		if !ready {
+			select {
+			case <-s.quitCh:
+				return
+			case <-time.After(100 * time.Millisecond):
+				s.RateLimitWarnf("Waiting for routing to established...")
+			}
+		} else {
+			break
+		}
+	}
+
 	for s.isRunning() {
 		switch n.State() {
 		case Follower:
@@ -1639,14 +1689,10 @@ func (n *raft) runAsFollower() {
 		case <-elect.C:
 			// If we are out of resources we just want to stay in this state for the moment.
 			if n.outOfResources() {
-				n.Lock()
-				n.resetElectionTimeout()
-				n.Unlock()
+				n.resetElectionTimeoutWithLock()
 				n.debug("Not switching to candidate, no resources")
 			} else if n.isObserver() {
-				n.Lock()
-				n.resetElect(48 * time.Hour)
-				n.Unlock()
+				n.resetElectWithLock(48 * time.Hour)
 				n.debug("Not switching to candidate, observer only")
 			} else if n.isCatchingUp() {
 				n.debug("Not switching to candidate, catching up")


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 99075a00..1feea011 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -4763,16 +4763,6 @@ func TestJetStreamClusterSuperClusterCrossClusterConsumerInterest(t *testing.T)
 	checkSubsPending(t, sub, 2)
 }
 
-func TestJetStreamNextReqFromMsg(t *testing.T) {
-	bef := time.Now()
-	expires, _, _, _, _, err := nextReqFromMsg([]byte(`{"expires":5000000000}`)) // nanoseconds
-	require_NoError(t, err)
-	now := time.Now()
-	if expires.Before(bef.Add(5*time.Second)) || expires.After(now.Add(5*time.Second)) {
-		t.Fatal("Expires out of expected range")
-	}
-}
-
 func TestJetStreamClusterSuperClusterPeerReassign(t *testing.T) {
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
@@ -5373,7 +5363,7 @@ func TestJetStreamCrossAccountMirrorsAndSources(t *testing.T) {
 
 }
 
-func TestJetStreamFailMirrorsAndSources(t *testing.T) {
+func TestJetStreamClusterFailMirrorsAndSources(t *testing.T) {
 	c := createJetStreamClusterWithTemplate(t, jsClusterMirrorSourceImportsTempl, "C1", 3)
 	defer c.shutdown()
 
@@ -8211,7 +8201,7 @@ func TestJetStreamClusterSourceAndMirrorConsumersLeaderChange(t *testing.T) {
 	})
 }
 
-func TestPurgeBySequence(t *testing.T) {
+func TestJetStreamClusterPurgeBySequence(t *testing.T) {
 	for _, st := range []StorageType{FileStorage, MemoryStorage} {
 		t.Run(st.String(), func(t *testing.T) {
 
@@ -8364,7 +8354,7 @@ func TestJetStreamClusterMaxConsumersMultipleConcurrentRequests(t *testing.T) {
 	}
 }
 
-func TestJetStreamPanicDecodingConsumerState(t *testing.T) {
+func TestJetStreamClusterPanicDecodingConsumerState(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
@@ -8449,7 +8439,7 @@ func TestJetStreamPanicDecodingConsumerState(t *testing.T) {
 }
 
 // Had a report of leaked subs with pull subscribers.
-func TestJetStreamPullConsumerLeakedSubs(t *testing.T) {
+func TestJetStreamClusterPullConsumerLeakedSubs(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
@@ -8510,7 +8500,7 @@ func TestJetStreamPullConsumerLeakedSubs(t *testing.T) {
 	}
 }
 
-func TestJetStreamPushConsumerQueueGroup(t *testing.T) {
+func TestJetStreamClusterPushConsumerQueueGroup(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
@@ -8754,7 +8744,7 @@ func TestJetStreamRaceOnRAFTCreate(t *testing.T) {
 	wg.Wait()
 }
 
-func TestJetStreamDeadlockOnVarz(t *testing.T) {
+func TestJetStreamClusterDeadlockOnVarz(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9430,7 +9420,7 @@ func TestJetStreamRollupSubjectAndWatchers(t *testing.T) {
 	expectUpdate("age", "50", 8)
 }
 
-func TestJetStreamAppendOnly(t *testing.T) {
+func TestJetStreamClusterAppendOnly(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
@@ -10674,7 +10664,7 @@ func TestJetStreamConsumerUpgrade(t *testing.T) {
 	t.Run("Clustered", func(t *testing.T) { testUpdate(t, c.randomServer()) })
 }
 
-func TestJetStreamAddConsumerWithInfo(t *testing.T) {
+func TestJetStreamClusterAddConsumerWithInfo(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	if config := s.JetStreamConfig(); config != nil {
 		defer removeDir(t, config.StoreDir)
@@ -11495,7 +11485,7 @@ func TestJetStreamConsumerDeliverNewBug(t *testing.T) {
 
 // If the config files have duplicate routes this can have the metagroup estimate a size for the system
 // which prevents reaching quorum and electing a meta-leader.
-func TestJetStreamDuplicateRoutesDisruptJetStreamMetaGroup(t *testing.T) {
+func TestJetStreamClusterDuplicateRoutesDisruptJetStreamMetaGroup(t *testing.T) {
 	tmpl := `
 	listen: 127.0.0.1:-1
 	server_name: %s
@@ -11534,7 +11524,7 @@ func TestJetStreamDuplicateRoutesDisruptJetStreamMetaGroup(t *testing.T) {
 	c.waitOnClusterReady()
 }
 
-func TestJetStreamDuplicateMsgIdsOnCatchupAndLeaderTakeover(t *testing.T) {
+func TestJetStreamClusterDuplicateMsgIdsOnCatchupAndLeaderTakeover(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
@@ -12158,6 +12148,17 @@ func TestJetStreamClusterMemoryConsumerInterestRetention(t *testing.T) {
 		m.AckSync()
 	}
 
+	checkFor(t, time.Second, 15*time.Millisecond, func() error {
+		si, err := js.StreamInfo("test")
+		if err != nil {
+			return err
+		}
+		if n := si.State.Msgs; n != 900 {
+			return fmt.Errorf("Waiting for msgs count to be 900, got %v", n)
+		}
+		return nil
+	})
+
 	si, err := js.StreamInfo("test")
 	require_NoError(t, err)
 
@@ -12189,7 +12190,7 @@ func TestJetStreamClusterMemoryConsumerInterestRetention(t *testing.T) {
 	}
 }
 
-func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
+func TestJetStreamClusterImportConsumerStreamSubjectRemap(t *testing.T) {
 	template := `
 	listen: 127.0.0.1:-1
 	server_name: %s
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 5a57b3a9..1b849746 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -645,6 +645,16 @@ func TestJetStreamConsumerMaxDeliveries(t *testing.T) {
 	}
 }
 
+func TestJetStreamNextReqFromMsg(t *testing.T) {
+	bef := time.Now()
+	expires, _, _, _, _, err := nextReqFromMsg([]byte(`{"expires":5000000000}`)) // nanoseconds
+	require_NoError(t, err)
+	now := time.Now()
+	if expires.Before(bef.Add(5*time.Second)) || expires.After(now.Add(5*time.Second)) {
+		t.Fatal("Expires out of expected range")
+	}
+}
+
 func TestJetStreamPullConsumerDelayedFirstPullWithReplayOriginal(t *testing.T) {
 	cases := []struct {
 		name    string
@@ -7241,7 +7251,7 @@ func TestJetStreamSuperClusterSystemLimitsPlacement(t *testing.T) {
 	}
 }
 
-func TestStreamLimitUpdate(t *testing.T) {
+func TestJetStreamStreamLimitUpdate(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	if config := s.JetStreamConfig(); config != nil {
 		defer removeDir(t, config.StoreDir)
@@ -15699,7 +15709,7 @@ func TestJetStreamPullConsumerHeartBeats(t *testing.T) {
 	}
 }
 
-func TestStorageReservedBytes(t *testing.T) {
+func TestJetStreamStorageReservedBytes(t *testing.T) {
 	const systemLimit = 1024
 	opts := DefaultTestOptions
 	opts.Port = -1
diff --git a/server/norace_test.go b/server/norace_test.go
index d34730e5..457532cd 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -3557,6 +3557,7 @@ func TestNoRaceJetStreamClusterStreamReset(t *testing.T) {
 	minElectionTimeout = 250 * time.Millisecond
 	maxElectionTimeout = time.Second
 	hbInterval = 50 * time.Millisecond
+	defer setDefaultRaftTimeouts()
 
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()