Changes to prevent deadlocks and loops on elections

lib/candidate.js

@@ -9,7 +9,6 @@ module.exports = Candidate;
 function Candidate(node, options) {
   State.call(this, node);
   this.options = options;
-  this.node.commonState.persisted.votedFor = this.node.id;
 
   this._startVoting();
 }
@@ -27,7 +26,7 @@ C._startVoting = function _startVoting() {
 
   async.series([
     startTimeout,
-    incrementTerm,
+    incrementTermAndVoteToSelf,
     requestVotes
   ]);
 
@@ -38,11 +37,14 @@ C._startVoting = function _startVoting() {
   }
 
   function onElectionTimeout() {
-    self.node.toState('candidate');
+    self.node.toState('candidate'); 
   }
 
-  function incrementTerm(cb) {
-    self.node.commonState.persisted.currentTerm += 1;
+  function incrementTermAndVoteToSelf(cb) {
+    self.node.currentTerm(self.node.currentTerm()+1);
+    self.node.commonState.persisted.votedFor = self.node.id;
+    self.node.commonState.persisted.voteTerm = self.node.currentTerm();
+
     cb();
   }
 
@@ -58,7 +60,7 @@ C._startVoting = function _startVoting() {
       }
 
       var args = {
-        term:         self.node.commonState.persisted.currentTerm,
+        term:         self.node.currentTerm(),
         candidateId:  self.node.id,
         lastLogIndex: self.node.commonState.persisted.log.length(),
         lastLogTerm:  lastLog && lastLog.term
@@ -69,10 +71,12 @@ C._startVoting = function _startVoting() {
     }
 
     function onBroadcastResponse(err, args) {
-      // TODO: what about the term update?
       if (args && args.voteGranted) {
         votedForMe ++;
         verifyMajority();
+      } else if (args && args.reason == 'too soon') {
+        // 'too soon' means that some one must have seen the leader alive
+        setImmediate(function () { self.node.toState('follower'); cb(); });
       }
     }
 

lib/default_node_options.js

@@ -1,5 +1,6 @@
 module.exports = {
   standby: false,
+  delayElectionTimeout: 1000,
   minElectionTimeout: 150,
   maxElectionTimeout: 300,
   heartbeatInterval: 50,

lib/follower.js

@@ -15,7 +15,8 @@ function Follower(node, options) {
   this.lastHeardFromLeader = undefined;
   this.lastLeader = undefined;
 
-  this.node.startElectionTimeout();
+  // delay the initial election timeout to allow the leader to send a heart beat
+  setTimeout(function () { self.node.startElectionTimeout(); }, options.delayElectionTimeout);
 
   this.node.commonState.persisted.votedFor = null;
 
@@ -72,7 +73,7 @@ F.onAppendEntries = function onAppendEntries(args, cb) {
     self.lastLeader = args.leaderId;
     self.lastHeardFromLeader = Date.now();
 
-    self.node.commonState.persisted.currentTerm = args.term;
+    self.node.currentTerm(args.term);
 
     self.node.commonState.volatile.commitIndex = Math.min(
       args.leaderCommit, self.node.commonState.persisted.log.length());

lib/leader.js

@@ -51,6 +51,10 @@ function Leader(node, options) {
     if (peer) {
       if (err) {
         self.emit('warning', err);
+        if (err.timeout) {
+          peer.meta.disconnect();
+          peer.meta.connect();
+        }
       }
       else if (args && args.term > self.node.currentTerm()) {
         self.node.currentTerm(args.term);

lib/node.js

@@ -309,7 +309,8 @@ N._join = function _join(peerDesc, metadata, connection) {
     peer.on('connected', onPeerConnected);
     peer.on('disconnected', onPeerDisconnected);
     peer.on('connecting', onPeerConnecting);
-
+    peer.on('error', onError);
+
     if (found) {
       self.emit('reconnected', peer);
     }
@@ -318,6 +319,10 @@ N._join = function _join(peerDesc, metadata, connection) {
     }
   }
 
+  function onError(err) {
+    self.emit('error', err);
+  }
+
   function onPeerCall(type, args, cb) {
     self.handlePeerCall(peer, type, args, cb);
   }

lib/peer.js

@@ -40,10 +40,12 @@ function Peer(id, metadata, options, connection, node) {
   }
 
   function invoke(work, done) {
+
     var calledback = false;
     self._invoke.call(self, work.type, work.args, callback);
 
     function callback() {
+
       if (!calledback) {
         calledback = true;
         work.cb.apply(null, arguments);
@@ -53,9 +55,11 @@ function Peer(id, metadata, options, connection, node) {
   }
 
   function receive(message, cb) {
+
     self.emit('call', message.type, message.args, callback);
 
     function callback() {
+
       message.cb.apply(null, arguments);
       cb();
     }
@@ -128,7 +132,7 @@ P._invoke = function _invoke(type, args, cb) {
   var timeout;
 
   if (this.node.options) {
-    setTimeout(onTimeout, this.node.options.commandTimeout);
+    timeout = setTimeout(onTimeout, this.node.options.commandTimeout);
   }
 
   this.emit('outgoing call', type, args);

lib/state.js

@@ -65,7 +65,7 @@ S.onRequestVote = function onRequestVote(args, cb) {
   if (args.term < currentTerm) {
     callback(false);
   }
-  else if (!state.votedFor || state.votedFor == args.candidateId) {
+  else if (!state.votedFor || (state.voteTerm < args.term) || state.votedFor == args.candidateId) {
     var lastLog = state.log.last();
     if (lastLog && lastLog.term < args.lastLogTerm) {
       callback(true);
@@ -84,6 +84,7 @@ S.onRequestVote = function onRequestVote(args, cb) {
   function callback(grant) {
     if (grant) {
       state.votedFor = args.candidateId;
+      state.voteTerm = args.term;
       self.node.emit('vote granted', state.votedFor);
     }
     cb(null, {term: currentTerm, voteGranted: grant});