Repository: nats-io/nats-server
Pull request #3041 (2022-04-15 20:25:57)

Raft improvements
We would declare lost quorum too quickly on scale up.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 658b2875..52e583b1 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1812,7 +1812,11 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 			// while the new members work through the election and catchup process.
 			// Double check for that instead of exiting here and being silent. e.g. nats stream update test --replicas=3
 			js.mu.RLock()
-			rg, ourID := sa.Group, cc.meta.ID()
+			rg := sa.Group
+			var ourID string
+			if cc.meta != nil {
+				ourID = cc.meta.ID()
+			}
 			bail := !rg.isMember(ourID)
 			if !bail {
 				// We know we are a member here, if this group is new and we are preferred allow us to answer.
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 047e30bc..ad4cb27e 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -773,12 +773,6 @@ func (cc *jetStreamCluster) isConsumerLeader(account, stream, consumer string) b
 	return false
 }
 
-// During recovery track any stream and consumer delete operations.
-type recoveryRemovals struct {
-	streams   map[string]*streamAssignment
-	consumers map[string]*consumerAssignment
-}
-
 func (js *jetStream) monitorCluster() {
 	s, n := js.server(), js.getMetaGroup()
 	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
@@ -822,11 +816,6 @@ func (js *jetStream) monitorCluster() {
 		}
 	}
 
-	rm := &recoveryRemovals{
-		streams:   make(map[string]*streamAssignment),
-		consumers: make(map[string]*consumerAssignment),
-	}
-
 	for {
 		select {
 		case <-s.quitCh:
@@ -839,21 +828,12 @@ func (js *jetStream) monitorCluster() {
 				if cei == nil {
 					// Signals we have replayed all of our metadata.
 					isRecovering = false
-					// Process any removes that are still valid after recovery.
-					for _, sa := range rm.streams {
-						js.processStreamRemoval(sa)
-					}
-					for _, ca := range rm.consumers {
-						js.processConsumerRemoval(ca)
-					}
-					// Clear.
-					rm = nil
 					s.Debugf("Recovered JetStream cluster metadata")
 					continue
 				}
 				ce := cei.(*CommittedEntry)
 				// FIXME(dlc) - Deal with errors.
-				if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering, rm); err == nil {
+				if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
 					_, nb := n.Applied(ce.Index)
 					if js.hasPeerEntries(ce.Entries) || didSnap || (didRemoval && time.Since(lastSnapTime) > 2*time.Second) {
 						// Since we received one make sure we have our own since we do not store
@@ -1345,7 +1325,7 @@ func (js *jetStream) hasPeerEntries(entries []*Entry) bool {
 	return false
 }
 
-func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *recoveryRemovals) (bool, bool, error) {
+func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool, bool, error) {
 	var didSnap, didRemove bool
 	for _, e := range entries {
 		if e.Type == EntrySnapshot {
@@ -1370,8 +1350,6 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *r
 				}
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
-					key := sa.Client.Account + ":" + sa.Config.Name
-					delete(rm.streams, key)
 				}
 				didRemove = js.processStreamAssignment(sa)
 			case removeStreamOp:
@@ -1382,12 +1360,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *r
 				}
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
-					key := sa.Client.Account + ":" + sa.Config.Name
-					rm.streams[key] = sa
-				} else {
-					js.processStreamRemoval(sa)
-					didRemove = true
 				}
+				js.processStreamRemoval(sa)
+				didRemove = true
 			case assignConsumerOp:
 				ca, err := decodeConsumerAssignment(buf[1:])
 				if err != nil {
@@ -1396,8 +1371,6 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *r
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.Client.Account + ":" + ca.Name
-					delete(rm.consumers, key)
 				}
 				js.processConsumerAssignment(ca)
 			case assignCompressedConsumerOp:
@@ -1408,8 +1381,6 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *r
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.Client.Account + ":" + ca.Name
-					delete(rm.consumers, key)
 				}
 				js.processConsumerAssignment(ca)
 			case removeConsumerOp:
@@ -1420,12 +1391,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool, rm *r
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.Client.Account + ":" + ca.Name
-					rm.consumers[key] = ca
-				} else {
-					js.processConsumerRemoval(ca)
-					didRemove = true
 				}
+				js.processConsumerRemoval(ca)
+				didRemove = true
 			case updateStreamOp:
 				sa, err := decodeStreamAssignment(buf[1:])
 				if err != nil {
@@ -2560,11 +2528,11 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 			js.mu.Unlock()
 		}
 
-		mset.setStreamAssignment(sa)
 		if err = mset.update(cfg); err != nil {
 			s.Warnf("JetStream cluster error updating stream %q for account %q: %v", cfg.Name, acc.Name, err)
-			mset.setStreamAssignment(osa)
 		}
+		// Set the new stream assignment.
+		mset.setStreamAssignment(sa)
 
 		// Make sure we are the leader now that we are R1.
 		if needsSetLeader {
diff --git a/server/raft.go b/server/raft.go
index ab297bd8..5c3bef6d 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -143,6 +143,7 @@ type raft struct {
 	elect    *time.Timer
 	active   time.Time
 	llqrt    time.Time
+	lsut     time.Time
 	term     uint64
 	pterm    uint64
 	pindex   uint64
@@ -227,12 +228,14 @@ type lps struct {
 }
 
 const (
-	minElectionTimeoutDefault = 2 * time.Second
-	maxElectionTimeoutDefault = 5 * time.Second
-	minCampaignTimeoutDefault = 100 * time.Millisecond
-	maxCampaignTimeoutDefault = 8 * minCampaignTimeoutDefault
-	hbIntervalDefault         = 500 * time.Millisecond
-	lostQuorumIntervalDefault = hbIntervalDefault * 20 // 10 seconds
+	minElectionTimeoutDefault      = 2 * time.Second
+	maxElectionTimeoutDefault      = 5 * time.Second
+	minCampaignTimeoutDefault      = 100 * time.Millisecond
+	maxCampaignTimeoutDefault      = 8 * minCampaignTimeoutDefault
+	hbIntervalDefault              = 500 * time.Millisecond
+	lostQuorumIntervalDefault      = hbIntervalDefault * 20 // 10 seconds
+	lostQuorumCheckIntervalDefault = hbIntervalDefault * 20 // 10 seconds
+
 )
 
 var (
@@ -242,6 +245,7 @@ var (
 	maxCampaignTimeout time.Duration
 	hbInterval         time.Duration
 	lostQuorumInterval time.Duration
+	lostQuorumCheck    time.Duration
 )
 
 func init() {
@@ -257,6 +261,7 @@ func setDefaultRaftTimeouts() {
 	maxCampaignTimeout = maxCampaignTimeoutDefault
 	hbInterval = hbIntervalDefault
 	lostQuorumInterval = lostQuorumIntervalDefault
+	lostQuorumCheck = lostQuorumCheckIntervalDefault
 }
 
 type RaftConfig struct {
@@ -1985,7 +1990,7 @@ func (n *raft) runAsLeader() {
 	hb := time.NewTicker(hbInterval)
 	defer hb.Stop()
 
-	lq := time.NewTicker(hbInterval * 2)
+	lq := time.NewTicker(lostQuorumCheck)
 	defer lq.Stop()
 
 	for {
@@ -2081,6 +2086,11 @@ func (n *raft) lostQuorum() bool {
 }
 
 func (n *raft) lostQuorumLocked() bool {
+	// Make sure we let any scale up actions settle before deciding.
+	if !n.lsut.IsZero() && time.Since(n.lsut) < lostQuorumInterval {
+		return false
+	}
+
 	now, nc := time.Now().UnixNano(), 1
 	for _, peer := range n.peers {
 		if now-peer.ts < int64(lostQuorumInterval) {
@@ -2474,6 +2484,7 @@ func (n *raft) adjustClusterSizeAndQuorum() {
 
 	if ncsz > pcsz {
 		n.debug("Expanding our clustersize: %d -> %d", pcsz, ncsz)
+		n.lsut = time.Now()
 	} else if ncsz < pcsz {
 		n.debug("Decreasing our clustersize: %d -> %d", pcsz, ncsz)
 		if n.state == Leader {
diff --git a/server/stream.go b/server/stream.go
index 51dab7c6..d2a2605d 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1466,7 +1466,7 @@ func (mset *stream) mirrorInfo() *StreamSourceInfo {
 	return mset.sourceInfo(mset.mirror)
 }
 
-const sourceHealthCheckInterval = 2 * time.Second
+const sourceHealthCheckInterval = 1 * time.Second
 
 // Will run as a Go routine to process mirror consumer messages.
 func (mset *stream) processMirrorMsgs() {


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index d5238146..fb18aacb 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3671,9 +3671,14 @@ func TestJetStreamClusterPeerOffline(t *testing.T) {
 
 func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 	// Make this shorter for the test.
-	old := lostQuorumInterval
+	oldInterval := lostQuorumInterval
 	lostQuorumInterval = hbIntervalDefault * 3
-	defer func() { lostQuorumInterval = old }()
+	oldCheck := lostQuorumCheck
+	lostQuorumCheck = hbIntervalDefault * 2
+	defer func() {
+		lostQuorumInterval = oldInterval
+		lostQuorumCheck = oldCheck
+	}()
 
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -12081,7 +12086,7 @@ func TestJetStreamClusterMovingStreamsWithMirror(t *testing.T) {
 	})
 	require_NoError(t, err)
 
-	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+	checkFor(t, 30*time.Second, 100*time.Millisecond, func() error {
 		si, err := js.StreamInfo("SOURCE")
 		if err != nil {
 			return err
@@ -12109,8 +12114,25 @@ func TestJetStreamClusterMovingStreamsWithMirror(t *testing.T) {
 	<-exited
 
 	if nnr := atomic.LoadUint64(&numNoResp); nnr > 0 {
-		t.Fatalf("Expected no failed message publishes, got %d", nnr)
+		if nnr > 5 {
+			t.Fatalf("Expected no or very few failed message publishes, got %d", nnr)
+		} else {
+			t.Logf("Got a few failed publishes: %d", nnr)
+		}
 	}
+
+	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("SOURCE")
+		require_NoError(t, err)
+		mi, err := js.StreamInfo("MIRROR")
+		require_NoError(t, err)
+
+		if si.State != mi.State {
+			return fmt.Errorf("Expected mirror to be the same, got %+v vs %+v", mi.State, si.State)
+		}
+		return nil
+	})
+
 }
 
 func TestJetStreamClusterMemoryConsumerInterestRetention(t *testing.T) {
@@ -12519,82 +12541,6 @@ func TestJetStreamClusterStreamAlternates(t *testing.T) {
 	getStreamInfo(nc, "C3")
 }
 
-func TestJetStreamClusterDeleteAndRestoreAndRestart(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "JSC", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST"})
-	require_NoError(t, err)
-
-	for i := 0; i < 10; i++ {
-		_, err := js.Publish("TEST", []byte("OK"))
-		require_NoError(t, err)
-	}
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy})
-	require_NoError(t, err)
-
-	require_NoError(t, js.DeleteConsumer("TEST", "dlc"))
-	require_NoError(t, js.DeleteStream("TEST"))
-
-	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST"})
-	require_NoError(t, err)
-
-	for i := 0; i < 22; i++ {
-		_, err := js.Publish("TEST", []byte("OK"))
-		require_NoError(t, err)
-	}
-	sub, err := js.SubscribeSync("TEST", nats.Durable("dlc"))
-	require_NoError(t, err)
-
-	for i := 0; i < 5; i++ {
-		m, err := sub.NextMsg(time.Second)
-		require_NoError(t, err)
-		m.AckSync()
-	}
-
-	// Now restart.
-	sl := c.streamLeader("$G", "TEST")
-	sl.Shutdown()
-	sl = c.restartServer(sl)
-	c.waitOnStreamLeader("$G", "TEST")
-
-	nc, js = jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	si, err := js.StreamInfo("TEST")
-	require_NoError(t, err)
-
-	if si.State.Msgs != 22 {
-		t.Fatalf("State is not correct after restart")
-	}
-
-	ci, err := js.ConsumerInfo("TEST", "dlc")
-	require_NoError(t, err)
-
-	if ci.AckFloor.Consumer != 5 {
-		t.Fatalf("Bad ack floor: %+v", ci.AckFloor)
-	}
-
-	// Now delete and make sure consumer does not come back.
-	// First add a delete something else.
-	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST2"})
-	require_NoError(t, err)
-	require_NoError(t, js.DeleteStream("TEST2"))
-	// Now the consumer.
-	require_NoError(t, js.DeleteConsumer("TEST", "dlc"))
-
-	sl.Shutdown()
-	c.restartServer(sl)
-	c.waitOnStreamLeader("$G", "TEST")
-
-	_, err = js.ConsumerInfo("TEST", "dlc")
-	require_Error(t, err)
-
-}
-
 // Support functions
 
 // Used to setup superclusters for tests.