Repository: nats-io/nats-server
Pull request #1843 (2021-01-26 04:27:42)

Fix for blocking on a channel for the raft impl, various test fixes for flappers.

/cc @nats-io/core



Patch:
diff --git a/server/raft.go b/server/raft.go
index 3f055101..d2ffd7e5 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -265,7 +265,7 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 		propc:    make(chan *Entry, 256),
 		applyc:   make(chan *CommittedEntry, 256),
 		leadc:    make(chan bool, 4),
-		stepdown: make(chan string, 4),
+		stepdown: make(chan string),
 	}
 	n.c.registerWithAccount(sacc)
 
@@ -1545,6 +1545,15 @@ func (n *raft) createCatchup(ae *appendEntry) string {
 	return inbox
 }
 
+// Attempt to stepdown, lock should be held.
+func (n *raft) attemptStepDown(newLeader string) {
+	select {
+	case n.stepdown <- newLeader:
+	default:
+		n.debug("Failed to place stepdown for new leader %q for %q", newLeader, n.group)
+	}
+}
+
 // processAppendEntry will process an appendEntry.
 func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	n.Lock()
@@ -1560,7 +1569,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		n.term = ae.term
 		n.vote = noVote
 		n.writeTermVote()
-		n.stepdown <- ae.leader
+		n.attemptStepDown(ae.leader)
 	}
 
 	// Catching up state.
@@ -1618,7 +1627,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		n.writeTermVote()
 		if n.state != Follower {
 			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.state, ae.leader)
-			n.stepdown <- ae.leader
+			n.attemptStepDown(ae.leader)
 		}
 	}
 
@@ -2006,13 +2015,11 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	// If this is a higher term go ahead and stepdown.
 	if vresp.term > n.term {
 		n.debug("Stepping down from candidate, detected higher term: %d vs %d", vresp.term, n.term)
-		stepdown := n.stepdown
 		n.term = vresp.term
 		n.vote = noVote
 		n.writeTermVote()
+		n.attemptStepDown(noLeader)
 		n.Unlock()
-
-		stepdown <- noLeader
 		n.sendReply(vr.reply, vresp.encode())
 		return nil
 	}


Test patch:
diff --git a/test/jetstream_cluster_test.go b/test/jetstream_cluster_test.go
index 19dd1f1d..53c4d3e0 100644
--- a/test/jetstream_cluster_test.go
+++ b/test/jetstream_cluster_test.go
@@ -687,9 +687,8 @@ func TestJetStreamClusterStreamSynchedTimeStamps(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 5)
 	defer c.shutdown()
 
-	s := c.randomServer()
-
 	// Client based API
+	s := c.randomServer()
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
@@ -720,6 +719,10 @@ func TestJetStreamClusterStreamSynchedTimeStamps(t *testing.T) {
 	sl.Shutdown()
 	c.waitOnNewStreamLeader("$G", "foo")
 
+	s = c.randomServer()
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
 	sub, err = js.SubscribeSync("foo")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -889,6 +892,9 @@ func TestJetStreamClusterStreamPublishWithActiveConsumers(t *testing.T) {
 		t.Fatalf("Consumer info did not match: %+v vs %+v", ci, ci2)
 	}
 
+	// In case the server above was also stream leader.
+	c.waitOnNewStreamLeader("$G", "foo")
+
 	// Now send more..
 	// Send 10 more messages.
 	for i := 11; i <= 20; i++ {
@@ -1446,11 +1452,19 @@ func TestJetStreamClusterExtendedStreamInfo(t *testing.T) {
 	if len(si.Cluster.Replicas) != 2 {
 		t.Fatalf("Expected %d replicas, got %d", 2, len(si.Cluster.Replicas))
 	}
-	for _, peer := range si.Cluster.Replicas {
-		if !peer.Current {
-			t.Fatalf("Expected replica to be current: %+v", peer)
+
+	// We may need to wait a bit for peers to catch up.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		for _, peer := range si.Cluster.Replicas {
+			if !peer.Current {
+				if si, err = js.StreamInfo("TEST"); err != nil {
+					t.Fatalf("Could not retrieve stream info")
+				}
+				return fmt.Errorf("Expected replica to be current: %+v", peer)
+			}
 		}
-	}
+		return nil
+	})
 
 	// Shutdown the leader.
 	oldLeader := c.streamLeader("$G", "TEST")
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index 419f74e2..43815488 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -4365,6 +4365,8 @@ func TestJetStreamSnapshotsAPI(t *testing.T) {
 	// Now connect through a cluster server and make sure we can get things to work this way as well.
 	nc2 := clientConnectToServer(t, ls)
 	defer nc2.Close()
+	// Wait a bit for interest to propagate.
+	time.Sleep(100 * time.Millisecond)
 
 	snapshot = snapshot[:0]
 
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index e14c18c6..e96057d6 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -656,7 +656,17 @@ func shutdownCluster(c *cluster) {
 }
 
 func (c *cluster) randomServer() *server.Server {
-	return c.servers[rand.Intn(len(c.servers))]
+	// Since these can be randomly shutdown in certain tests make sure they are running first.
+	// Copy our servers list and shuffle then walk looking for first running server.
+	cs := append(c.servers[:0:0], c.servers...)
+	rand.Shuffle(len(cs), func(i, j int) { cs[i], cs[j] = cs[j], cs[i] })
+
+	for _, s := range cs {
+		if s.Running() {
+			return s
+		}
+	}
+	return nil
 }
 
 func (c *cluster) totalSubs() int {