[IMPROVED] Raft layer improvements (#4020)

Two changes here.

1. Revert of a change regarding handling of leadership transfer entries
that caused instability in some tests and a bad upgrade experience.

2. Make sure if we detect that a stream or a consumer is already running
the monitoring routines that we do not stop the underlying raft group
since it will be being used in the monitor routine and would stall that
asset.

Signed-off-by: Derek Collison <derek@nats.io>

server/jetstream_cluster.go

@@ -436,11 +436,9 @@ func (cc *jetStreamCluster) isStreamCurrent(account, stream string) bool {
 
 // isStreamHealthy will determine if the stream is up to date or very close.
 // For R1 it will make sure the stream is present on this server.
+// Read lock should be held.
 func (js *jetStream) isStreamHealthy(account, stream string) bool {
-	js.mu.RLock()
-	defer js.mu.RUnlock()
 	cc := js.cluster
-
 	if cc == nil {
 		// Non-clustered mode
 		return true
@@ -480,11 +478,9 @@ func (js *jetStream) isStreamHealthy(account, stream string) bool {
 
 // isConsumerCurrent will determine if the consumer is up to date.
 // For R1 it will make sure the consunmer is present on this server.
+// Read lock should be held.
 func (js *jetStream) isConsumerCurrent(account, stream, consumer string) bool {
-	js.mu.RLock()
-	defer js.mu.RUnlock()
 	cc := js.cluster
-
 	if cc == nil {
 		// Non-clustered mode
 		return true
@@ -1943,9 +1939,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		return
 	}
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	defer n.Stop()
-
 	// Make sure only one is running.
 	if mset != nil {
 		if mset.checkInMonitor() {
@@ -1954,6 +1947,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		defer mset.clearMonitorRunning()
 	}
 
+	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
+	// This should be below the checkInMonitor call though to avoid stopping it out
+	// from underneath the one that is running since it will be the same raft node.
+	defer n.Stop()
+
 	qch, lch, aq, uch, ourPeerId := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), mset.updateC(), meta.ID()
 
 	s.Debugf("Starting stream monitor for '%s > %s' [%s]", sa.Client.serviceAccount(), sa.Config.Name, n.Group())
@@ -4187,15 +4185,17 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		return
 	}
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	defer n.Stop()
-
 	// Make sure only one is running.
 	if o.checkInMonitor() {
 		return
 	}
 	defer o.clearMonitorRunning()
 
+	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
+	// This should be below the checkInMonitor call though to avoid stopping it out
+	// from underneath the one that is running since it will be the same raft node.
+	defer n.Stop()
+
 	qch, lch, aq, uch, ourPeerId := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), o.updateC(), cc.meta.ID()
 
 	s.Debugf("Starting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())

server/jetstream_cluster_3_test.go

@@ -2242,14 +2242,15 @@ func TestJetStreamClusterMemLeaderRestart(t *testing.T) {
 
 	// Make sure that we have a META leader (there can always be a re-election)
 	c.waitOnLeader()
+	c.waitOnStreamLeader(globalAccountName, "foo")
 
 	// Should still have quorum and a new leader
-	checkFor(t, time.Second, 200*time.Millisecond, func() error {
+	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
 		osi, err = jsc.StreamInfo("foo")
 		if err != nil {
 			return fmt.Errorf("expected healthy stream asset, got %s", err.Error())
 		}
-		if osi.Cluster.Leader == "" {
+		if osi.Cluster.Leader == _EMPTY_ {
 			return fmt.Errorf("expected healthy stream asset with new leader")
 		}
 		if osi.State.Msgs != uint64(toSend) {

server/monitor.go

@@ -3125,6 +3125,11 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	// Range across all accounts, the streams assigned to them, and the consumers.
 	// If they are assigned to this server check their status.
 	ourID := meta.ID()
+
+	// TODO(dlc) - Might be better here to not hold the lock the whole time.
+	js.mu.RLock()
+	defer js.mu.RUnlock()
+
 	for acc, asa := range cc.streams {
 		for stream, sa := range asa {
 			if sa.Group.isMember(ourID) {

server/raft.go

@@ -2511,13 +2511,6 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// We pass these up as well.
 			committed = append(committed, e)
-
-		case EntryLeaderTransfer:
-			if n.state == Leader {
-				n.debug("Stepping down")
-				n.stepdown.push(noLeader)
-			}
-			// No-op
 		}
 	}
 	// Pass to the upper layers if we have normal entries.
@@ -3056,6 +3049,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	for _, e := range ae.entries {
 		switch e.Type {
 		case EntryLeaderTransfer:
+			// Only process these if they are new, so no replays or catchups.
 			if isNew {
 				maybeLeader := string(e.Data)
 				if maybeLeader == n.id && !n.observer && !n.paused {
@@ -3156,7 +3150,7 @@ func (n *raft) buildAppendEntry(entries []*Entry) *appendEntry {
 
 // Determine if we should store an entry.
 func (ae *appendEntry) shouldStore() bool {
-	return ae != nil && len(ae.entries) > 0 && ae.entries[0].Type != EntryLeaderTransfer
+	return ae != nil && len(ae.entries) > 0
 }
 
 // Store our append entry to our WAL.