Add external state support

Add support for external state (log) to influence leader voting. This,
in effect, implements lastLogIndex and lastLogTerm sent on RequestVote
RPCs from the Raft paper. This works by exposing two callbacks: one that
calls into the user on RequestVote to get the candidate's state and one
that calls into the user upon receiving a RequestVote to determine if a
vote should be granted based on comparing the logs.

From Raft:
1. Reply false if term < currentTerm (§5.1)
2. If votedFor is null or candidateId, and candidate’s log is at
least as up-to-date as receiver’s log, grant vote (§5.2, §5.4)

chan_handler.go

@@ -5,6 +5,8 @@ package graft
 // ChanHandler is a convenience handler when a user wants to simply use
 // channels for the async handling of errors and state changes.
 type ChanHandler struct {
+	LogPositionHandler
+
 	// Chan to receive state changes.
 	stateChangeChan chan<- StateChange
 	// Chan to receive errors.
@@ -20,10 +22,11 @@ type StateChange struct {
 	To State
 }
 
-func NewChanHandler(scCh chan<- StateChange, errCh chan<- error) *ChanHandler {
+func NewChanHandler(logHandler LogPositionHandler, scCh chan<- StateChange, errCh chan<- error) *ChanHandler {
 	return &ChanHandler{
-		stateChangeChan: scCh,
-		errorChan:       errCh,
+		LogPositionHandler: logHandler,
+		stateChangeChan:    scCh,
+		errorChan:          errCh,
 	}
 }
 

handler_test.go

@@ -1,8 +1,9 @@
-// Copyright 2013-2016 Apcera Inc. All rights reserved.
+// Copyright 2013-2017 Apcera Inc. All rights reserved.
 
 package graft
 
 import (
+	"encoding/binary"
 	"fmt"
 	"os"
 	"strconv"
@@ -12,6 +13,20 @@ import (
 	"github.com/nats-io/graft/pb"
 )
 
+type logPositionHandler struct {
+	logIndex uint32
+}
+
+func (l *logPositionHandler) CurrentLogPosition() []byte {
+	buf := make([]byte, 4)
+	binary.BigEndian.PutUint32(buf, l.logIndex)
+	return buf
+}
+
+func (l *logPositionHandler) GrantVote(position []byte) bool {
+	return binary.BigEndian.Uint32(position) >= l.logIndex
+}
+
 // Dumb wait program to sync on callbacks, etc... Will timeout
 func wait(t *testing.T, ch chan StateChange) *StateChange {
 	select {
@@ -43,7 +58,8 @@ func TestStateChangeHandler(t *testing.T) {
 	// Use ChanHandler
 	scCh := make(chan StateChange)
 	errCh := make(chan error)
-	chHand := NewChanHandler(scCh, errCh)
+	lpHandler := &logPositionHandler{}
+	chHand := NewChanHandler(lpHandler, scCh, errCh)
 
 	node, err := New(ci, chHand, rpc, log)
 	if err != nil {
@@ -78,7 +94,8 @@ func TestErrorHandler(t *testing.T) {
 	// Use ChanHandler
 	scCh := make(chan StateChange)
 	errCh := make(chan error)
-	chHand := NewChanHandler(scCh, errCh)
+	lpHandler := &logPositionHandler{}
+	chHand := NewChanHandler(lpHandler, scCh, errCh)
 
 	node, err := New(ci, chHand, rpc, log)
 	if err != nil {
@@ -111,7 +128,8 @@ func TestChandHandlerNotBlockingNode(t *testing.T) {
 	// Use ChanHandler
 	scCh := make(chan StateChange)
 	errCh := make(chan error)
-	chHand := NewChanHandler(scCh, errCh)
+	lpHandler := &logPositionHandler{}
+	chHand := NewChanHandler(lpHandler, scCh, errCh)
 
 	node, err := New(ci, chHand, rpc, log)
 	if err != nil {

node.go

@@ -1,4 +1,6 @@
-// Copyright 2013-2016 Apcera Inc. All rights reserved.
+// Copyright 2013-2017 Apcera Inc. All rights reserved.
+
+//go:generate protoc -I=. -I=$GOPATH/src --gofast_out=. ./pb/protocol.proto
 
 // Graft is a RAFT implementation.
 // Currently only the election functionality is supported.
@@ -78,8 +80,24 @@ type ClusterInfo struct {
 	Size int
 }
 
+// LogPositionHandler is used to interrogate the state of the log.
+type LogPositionHandler interface {
+	// CurrentLogPosition returns an opaque byte slice that represents the
+	// current position in the node's log.
+	CurrentLogPosition() []byte
+
+	// GrantVote is called when a candidate peer has requested a vote. The
+	// peer's log position is passed as an opaque byte slice as returned by
+	// CurrentLogPosition. The returned bool determines if the vote should be
+	// granted because the candidate's log is at least as up-to-date as the
+	// receiver's log.
+	GrantVote(position []byte) bool
+}
+
 // A Handler can process async callbacks from a Graft node.
 type Handler interface {
+	LogPositionHandler
+
 	// Process async errors that are encountered by the node.
 	AsyncError(error)
 
@@ -247,8 +265,9 @@ func (n *Node) runAsCandidate() {
 
 	// Initiate an Election
 	vreq := &pb.VoteRequest{
-		Term:      n.term,
-		Candidate: n.id,
+		Term:        n.term,
+		Candidate:   n.id,
+		LogPosition: n.handler.CurrentLogPosition(),
 	}
 	// Collect the votes.
 	// We will vote for ourselves, so start at 1.
@@ -439,15 +458,12 @@ func (n *Node) handleVoteRequest(vreq *pb.VoteRequest) bool {
 
 	deny := &pb.VoteResponse{Term: n.term, Granted: false}
 
-	// Old term, reject
-	if vreq.Term < n.term {
+	// Old term or candidate's log is behind, reject
+	if vreq.Term < n.term || !n.handler.GrantVote(vreq.LogPosition) {
 		n.rpc.SendVoteResponse(vreq.Candidate, deny)
 		return false
 	}
 
-	// Save state flag
-	saveState := false
-
 	// This will trigger a return from the current runAs loop.
 	stepDown := false
 
@@ -457,7 +473,6 @@ func (n *Node) handleVoteRequest(vreq *pb.VoteRequest) bool {
 		n.vote = NO_VOTE
 		n.leader = NO_LEADER
 		stepDown = true
-		saveState = true
 	}
 
 	// If we are the Leader, deny request unless we have seen
@@ -477,17 +492,15 @@ func (n *Node) handleVoteRequest(vreq *pb.VoteRequest) bool {
 
 	n.setVote(vreq.Candidate)
 
-	// Write our state if needed.
-	if saveState {
-		if err := n.writeState(); err != nil {
-			// We have failed to update our state. Process the error
-			// and deny the vote.
-			n.handleError(err)
-			n.setVote(NO_VOTE)
-			n.rpc.SendVoteResponse(vreq.Candidate, deny)
-			n.resetElectionTimeout()
-			return true
-		}
+	// Write our state.
+	if err := n.writeState(); err != nil {
+		// We have failed to update our state. Process the error
+		// and deny the vote.
+		n.handleError(err)
+		n.setVote(NO_VOTE)
+		n.rpc.SendVoteResponse(vreq.Candidate, deny)
+		n.resetElectionTimeout()
+		return true
 	}
 
 	// Send our acceptance.