More robust testing, TestSplitVotes

candidate.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net/http"
 	"sync"
 	"time"
@@ -34,9 +35,7 @@ type CandidateState int
 const (
 	StateLeader CandidateState = iota
 	StateNotLeader
-)
 
-const (
 	maxVotePeriod  = 5 * time.Second
 	voteTimeout    = 10 * time.Second
 	leadershipWait = 15 * time.Second
@@ -44,6 +43,11 @@ const (
 	maxFastVotePeriod = 100 * time.Millisecond
 )
 
+var StateName = map[CandidateState]string{
+	StateLeader:    "LEADER",
+	StateNotLeader: "NOT_LEADER",
+}
+
 func NewCandidate(numVoters int, signingKey string) *Candidate {
 	change := make(chan CandidateState, 100)
 
@@ -182,12 +186,23 @@ func (c *Candidate) elect(v *vote) {
 	defer c.mu.Unlock()
 
 	state := StateNotLeader
+	no := 0
+	yes := 0
 
 	defer func() {
 		if c.state == state {
 			return
 		}
 
+		log.Printf(
+			"[elect] transitioning %s -> %s (no=%d yes=%d min_yes=%d)",
+			StateName[c.state],
+			StateName[state],
+			no,
+			yes,
+			c.numVoters/2+1,
+		)
+
 		c.state = state
 		c.c <- state
 	}()
@@ -197,9 +212,6 @@ func (c *Candidate) elect(v *vote) {
 		c.votes[v.VoterID] = v
 	}
 
-	no := 0
-	yes := 0
-
 	for key, vote := range c.votes {
 		if time.Since(vote.received) > voteTimeout {
 			// Remove stale vote from consideration
@@ -220,7 +232,7 @@ func (c *Candidate) elect(v *vote) {
 		yes++
 	}
 
-	if no > 0 || yes <= c.numVoters/2 {
+	if no > 0 || yes < c.numVoters/2+1 {
 		// We lost the vote
 		c.firstYes = time.Time{}
 		state = StateNotLeader

elect_test.go

@@ -10,7 +10,7 @@ import (
 func TestOne(t *testing.T) {
 	t.Parallel()
 
-	ts := NewTestSystem(t, 1)
+	ts := NewTestSystem(t, 1, 1)
 	defer ts.Stop()
 
 	require.False(t, ts.Candidate(0).IsLeader())
@@ -21,49 +21,66 @@ func TestOne(t *testing.T) {
 func TestThree(t *testing.T) {
 	t.Parallel()
 
-	ts := NewTestSystem(t, 3)
+	ts := NewTestSystem(t, 3, 3)
 	defer ts.Stop()
 
 	require.False(t, ts.Candidate(0).IsLeader())
 	require.False(t, ts.Candidate(1).IsLeader())
 	require.False(t, ts.Candidate(2).IsLeader())
 
-	require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond)
+	{
-	require.False(t, ts.Candidate(1).IsLeader())
+		w := NewWaiter()
-	require.False(t, ts.Candidate(2).IsLeader())
+
+		w.Async(func() { require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(2).IsLeader, 20*time.Second, 100*time.Millisecond) })
+
+		w.Wait()
+	}
 }
 
 func TestFailover(t *testing.T) {
 	t.Parallel()
 
-	ts := NewTestSystem(t, 3)
+	ts := NewTestSystem(t, 3, 3)
 	defer ts.Stop()
 
 	require.False(t, ts.Candidate(0).IsLeader())
 	require.False(t, ts.Candidate(1).IsLeader())
 	require.False(t, ts.Candidate(2).IsLeader())
 
-	require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond)
+	{
-	require.False(t, ts.Candidate(1).IsLeader())
+		w := NewWaiter()
-	require.False(t, ts.Candidate(2).IsLeader())
+
+		w.Async(func() { require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(2).IsLeader, 20*time.Second, 100*time.Millisecond) })
+
+		w.Wait()
+	}
 
 	ts.SetServer(1)
 
 	require.Eventually(t, func() bool { return !ts.Candidate(0).IsLeader() }, 15*time.Second, 100*time.Millisecond)
-
 	// New candidate must not get leadership before old candidate loses it
 	require.False(t, ts.Candidate(1).IsLeader())
 	require.False(t, ts.Candidate(2).IsLeader())
 
-	require.Eventually(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond)
+	{
-	require.False(t, ts.Candidate(0).IsLeader())
+		w := NewWaiter()
-	require.False(t, ts.Candidate(2).IsLeader())
+
+		w.Async(func() { require.Eventually(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(2).IsLeader, 20*time.Second, 100*time.Millisecond) })
+
+		w.Wait()
+	}
 }
 
 func TestPartialVotes(t *testing.T) {
 	t.Parallel()
 
-	ts := NewTestSystem(t, 3)
+	ts := NewTestSystem(t, 3, 3)
 	defer ts.Stop()
 
 	ts.Proxy(0).SetRefuse(true)
@@ -72,7 +89,37 @@ func TestPartialVotes(t *testing.T) {
 	require.False(t, ts.Candidate(1).IsLeader())
 	require.False(t, ts.Candidate(2).IsLeader())
 
-	require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond)
+	{
+		w := NewWaiter()
+
+		w.Async(func() { require.Eventually(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(2).IsLeader, 20*time.Second, 100*time.Millisecond) })
+
+		w.Wait()
+	}
+}
+
+func TestSplitVotes(t *testing.T) {
+	t.Parallel()
+
+	ts := NewTestSystem(t, 3, 3)
+	defer ts.Stop()
+
+	ts.SetServerForVoter(1, 1)
+	ts.SetServerForVoter(1, 2)
+
+	require.False(t, ts.Candidate(0).IsLeader())
 	require.False(t, ts.Candidate(1).IsLeader())
 	require.False(t, ts.Candidate(2).IsLeader())
+
+	{
+		w := NewWaiter()
+
+		w.Async(func() { require.Eventually(t, ts.Candidate(1).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(0).IsLeader, 20*time.Second, 100*time.Millisecond) })
+		w.Async(func() { require.Never(t, ts.Candidate(2).IsLeader, 20*time.Second, 100*time.Millisecond) })
+
+		w.Wait()
+	}
 }

lib_test.go

@@ -27,9 +27,13 @@ type TestSystem struct {
 	proxies    []*proxy.Proxy
 }
 
-func NewTestServer(t *testing.T, signingKey string) *TestServer {
+type Waiter struct {
+	chans []<-chan bool
+}
+
+func NewTestServer(t *testing.T, numVoters int, signingKey string) *TestServer {
 	ts := &TestServer{
-		Candidate: elect.NewCandidate(1, signingKey),
+		Candidate: elect.NewCandidate(numVoters, signingKey),
 		listener:  lo.Must(net.ListenTCP("tcp", nil)),
 	}
 
@@ -55,13 +59,16 @@ func (ts *TestServer) Addr() *net.TCPAddr {
 	return ts.listener.Addr().(*net.TCPAddr)
 }
 
-func NewTestSystem(t *testing.T, num int) *TestSystem {
+func NewTestSystem(t *testing.T, numCandidates, numVoters int) *TestSystem {
 	ts := &TestSystem{
 		signingKey: uniuri.New(),
 	}
 
-	for i := 0; i < num; i++ {
+	for i := 0; i < numCandidates; i++ {
-		ts.servers = append(ts.servers, NewTestServer(t, ts.signingKey))
+		ts.servers = append(ts.servers, NewTestServer(t, numVoters, ts.signingKey))
+	}
+
+	for i := 0; i < numVoters; i++ {
 		ts.proxies = append(ts.proxies, proxy.NewProxy(t, ts.Server(0).Addr()))
 		ts.voters = append(ts.voters, elect.NewVoter(ts.Proxy(i).HTTP(), ts.signingKey))
 	}
@@ -83,12 +90,16 @@ func (ts *TestSystem) Stop() {
 	}
 }
 
-func (ts *TestSystem) SetServer(i int) {
+func (ts *TestSystem) SetServer(server int) {
 	for _, p := range ts.proxies {
-		p.SetBackend(ts.Server(i).Addr())
+		p.SetBackend(ts.Server(server).Addr())
 	}
 }
 
+func (ts *TestSystem) SetServerForVoter(server, voter int) {
+	ts.Proxy(voter).SetBackend(ts.Server(server).Addr())
+}
+
 func (ts *TestSystem) Candidate(i int) *elect.Candidate {
 	return ts.servers[i].Candidate
 }
@@ -104,3 +115,23 @@ func (ts *TestSystem) Server(i int) *TestServer {
 func (ts *TestSystem) Voter(i int) *elect.Voter {
 	return ts.voters[i]
 }
+
+func NewWaiter() *Waiter {
+	return &Waiter{}
+}
+
+func (w *Waiter) Wait() {
+	for _, ch := range w.chans {
+		<-ch
+	}
+}
+
+func (w *Waiter) Async(cb func()) {
+	ch := make(chan bool)
+	w.chans = append(w.chans, ch)
+
+	go func() {
+		defer close(ch)
+		cb()
+	}()
+}