Add external state support

chan_handler.go

@@ -1,10 +1,12 @@
-// Copyright 2013-2016 Apcera Inc. All rights reserved.
+// Copyright 2013-2017 Apcera Inc. All rights reserved.
 
 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 {
+	StateMachineHandler
+
 	// Chan to receive state changes.
 	stateChangeChan chan<- StateChange
 	// Chan to receive errors.
@@ -20,10 +22,24 @@ type StateChange struct {
 	To State
 }
 
+// NewChanHandler returns a Handler implementation which uses channels for
+// handling errors and state changes.
 func NewChanHandler(scCh chan<- StateChange, errCh chan<- error) *ChanHandler {
+	return NewChanHandlerWithStateMachine(new(defaultStateMachineHandler), scCh, errCh)
+}
+
+// NewChanHandlerWithStateMachine returns a Handler implementation which uses
+// channels for handling errors and state changes and a StateMachineHandler for
+// hooking into external state. The external state machine influences leader
+// election votes.
+func NewChanHandlerWithStateMachine(
+	stateHandler StateMachineHandler,
+	scCh chan<- StateChange,
+	errCh chan<- error) *ChanHandler {
 	return &ChanHandler{
-		stateChangeChan: scCh,
-		errorChan:       errCh,
+		StateMachineHandler: stateHandler,
+		stateChangeChan:     scCh,
+		errorChan:           errCh,
 	}
 }
 

handler_test.go

@@ -1,4 +1,4 @@
-// Copyright 2013-2016 Apcera Inc. All rights reserved.
+// Copyright 2013-2017 Apcera Inc. All rights reserved.
 
 package graft
 

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,38 @@ type ClusterInfo struct {
 	Size int
 }
 
+// StateMachineHandler is used to interrogate an external state machine.
+type StateMachineHandler interface {
+	// CurrentState returns an opaque byte slice that represents the current
+	// state of the state machine.
+	CurrentState() []byte
+
+	// GrantVote is called when a candidate peer has requested a vote. The
+	// peer's state machine position is passed as an opaque byte slice as
+	// returned by CurrentState. The returned bool determines if the vote
+	// should be granted because the candidate's state machine is at least as
+	// up-to-date as the receiver's state machine.
+	GrantVote(position []byte) bool
+}
+
+// defaultStateMachineHandler implements the StateMachineHandler interface by
+// always granting a vote.
+type defaultStateMachineHandler struct{}
+
+// CurrentState returns nil for default behavior.
+func (d *defaultStateMachineHandler) CurrentState() []byte {
+	return nil
+}
+
+// GrantVote always returns true for default behavior.
+func (d *defaultStateMachineHandler) GrantVote(position []byte) bool {
+	return true
+}
+
 // A Handler can process async callbacks from a Graft node.
 type Handler interface {
+	StateMachineHandler
+
 	// Process async errors that are encountered by the node.
 	AsyncError(error)
 
@@ -247,8 +279,9 @@ func (n *Node) runAsCandidate() {
 
 	// Initiate an Election
 	vreq := &pb.VoteRequest{
-		Term:      n.term,
-		Candidate: n.id,
+		Term:         n.term,
+		Candidate:    n.id,
+		CurrentState: n.handler.CurrentState(),
 	}
 	// Collect the votes.
 	// We will vote for ourselves, so start at 1.
@@ -439,15 +472,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.CurrentState) {
 		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 +487,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 +506,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.