Merge pull request #1663 from influxdb/raft

Fix candidate step down on heartbeat

raft/clock.go

@@ -43,7 +43,8 @@ func (c *Clock) AfterApplyInterval() <-chan chan struct{} { return newClockChan(
 // AfterElectionTimeout returns a channel that fires after a duration that is
 // between the election timeout and double the election timeout.
 func (c *Clock) AfterElectionTimeout() <-chan chan struct{} {
-	return newClockChan(c.ElectionTimeout + time.Duration(rand.Intn(int(c.ElectionTimeout))))
+	d := c.ElectionTimeout + time.Duration(rand.Intn(int(c.ElectionTimeout)))
+	return newClockChan(d)
 }
 
 // AfterHeartbeatInterval returns a channel that fires after the heartbeat interval.

raft/internal_test.go

@@ -1,5 +1,73 @@
 package raft
 
+import (
+	"encoding/binary"
+	"io"
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+func TestLog_followerLoop(t *testing.T) {
+	l := NewInitializedLog()
+	defer CloseLog(l)
+}
+
 func (l *Log) WaitUncommitted(index uint64) error { return l.waitUncommitted(index) }
 func (l *Log) WaitCommitted(index uint64) error   { return l.waitCommitted(index) }
 func (l *Log) WaitApplied(index uint64) error     { return l.Wait(index) }
+
+// NewOpenedLog returns an opened Log. Panic on error.
+func NewOpenedLog() *Log {
+	l := NewLog()
+	l.FSM = &IndexFSM{}
+	if err := l.Open(tempfile()); err != nil {
+		panic(err.Error())
+	}
+	return l
+}
+
+// NewInitializedLog returns an opened & initialized Log. Panic on error.
+func NewInitializedLog() *Log {
+	l := NewOpenedLog()
+	if err := l.Initialize(); err != nil {
+		panic(err.Error())
+	}
+	return l
+}
+
+// CloseLog closes a log and deletes its underlying path.
+func CloseLog(l *Log) {
+	defer os.RemoveAll(l.Path())
+	l.Close()
+}
+
+// IndexFSM represents a state machine that only records the last applied index.
+type IndexFSM struct {
+	index uint64
+}
+
+// MustApply updates the index.
+func (fsm *IndexFSM) MustApply(entry *LogEntry) { fsm.index = entry.Index }
+
+// Index returns the highest applied index.
+func (fsm *IndexFSM) Index() (uint64, error) { return fsm.index, nil }
+
+// Snapshot writes the FSM's index as the snapshot.
+func (fsm *IndexFSM) Snapshot(w io.Writer) (uint64, error) {
+	return fsm.index, binary.Write(w, binary.BigEndian, fsm.index)
+}
+
+// Restore reads the snapshot from the reader.
+func (fsm *IndexFSM) Restore(r io.Reader) error {
+	return binary.Read(r, binary.BigEndian, &fsm.index)
+}
+
+// tempfile returns the path to a non-existent file in the temp directory.
+func tempfile() string {
+	f, _ := ioutil.TempFile("", "raft-")
+	path := f.Name()
+	f.Close()
+	os.Remove(path)
+	return path
+}

raft/log.go

@@ -788,7 +788,7 @@ func (l *Log) candidateLoop(closing <-chan struct{}) State {
 			return Stopped
 		case hb := <-l.heartbeats:
 			l.mu.Lock()
-			if hb.term > l.term {
+			if hb.term >= l.term {
 				l.term = hb.term
 				l.votedFor = 0
 				l.leaderID = hb.leaderID
@@ -1386,7 +1386,7 @@ func (l *Log) RequestVote(term, candidateID, lastLogIndex, lastLogTerm uint64) (
 		return ErrAlreadyVoted
 	} else if lastLogTerm < l.lastLogTerm {
 		return ErrOutOfDateLog
-	} else if lastLogTerm == l.term && lastLogIndex < l.lastLogIndex {
+	} else if lastLogTerm == l.lastLogTerm && lastLogIndex < l.lastLogIndex {
 		return ErrOutOfDateLog
 	}
 

raft/log_test.go

@@ -297,8 +297,9 @@ func TestCluster_Elect_RealTime(t *testing.T) {
 
 	// Verify FSM indicies match.
 	for i, l := range c.Logs {
-		if exp, fsm := commandN+minIndex, l.FSM.(*IndexFSM); exp != fsm.index {
-			t.Errorf("fsm index mismatch(%d): exp=%d, got=%d", i, exp, fsm.index)
+		fsmIndex, _ := l.FSM.(*raft.IndexFSM).Index()
+		if exp := commandN + minIndex; exp != fsmIndex {
+			t.Errorf("fsm index mismatch(%d): exp=%d, got=%d", i, exp, fsmIndex)
 		}
 	}
 }
@@ -333,35 +334,13 @@ func benchmarkClusterApply(b *testing.B, logN int) {
 
 	// Verify FSM indicies match.
 	for i, l := range c.Logs {
-		if fsm := l.FSM.(*IndexFSM); index != fsm.index {
-			b.Errorf("fsm index mismatch(%d): exp=%d, got=%d", i, index, fsm.index)
+		fsmIndex, _ := l.FSM.(*raft.IndexFSM).Index()
+		if index != fsmIndex {
+			b.Errorf("fsm index mismatch(%d): exp=%d, got=%d", i, index, fsmIndex)
 		}
 	}
 }
 
-// IndexFSM represents a state machine that only records the last applied index.
-type IndexFSM struct {
-	index uint64
-}
-
-// MustApply updates the index.
-func (fsm *IndexFSM) MustApply(entry *raft.LogEntry) { fsm.index = entry.Index }
-
-// Index returns the highest applied index.
-func (fsm *IndexFSM) Index() (uint64, error) { return fsm.index, nil }
-
-// Snapshot writes the FSM's index as the snapshot.
-func (fsm *IndexFSM) Snapshot(w io.Writer) (uint64, error) {
-	return fsm.index, binary.Write(w, binary.BigEndian, fsm.index)
-}
-
-// Restore reads the snapshot from the reader.
-func (fsm *IndexFSM) Restore(r io.Reader) error {
-	return binary.Read(r, binary.BigEndian, &fsm.index)
-}
-
-func indexFSMFunc() raft.FSM { return &IndexFSM{} }
-
 // Cluster represents a collection of nodes that share the same mock clock.
 type Cluster struct {
 	Logs []*Log
@@ -643,7 +622,8 @@ func (fsm *FSM) Restore(r io.Reader) error {
 	return nil
 }
 
-func fsmFunc() raft.FSM { return &FSM{} }
+func fsmFunc() raft.FSM      { return &FSM{} }
+func indexFSMFunc() raft.FSM { return &raft.IndexFSM{} }
 
 // seq implements the raft.Log#Rand interface and returns incrementing ints.
 func seq() func() int64 {