Merge pull request #32 from xiangli-cmu/master

gofmt and add func to get the committed index

log.go

@@ -326,7 +326,7 @@ func (l *Log) SetCommitIndex(index uint64) error {
 
 	// Do not allow previous indices to be committed again.
 
-	// This could happens, since the guarantee is that the new leader has up-to-dated 
+	// This could happens, since the guarantee is that the new leader has up-to-dated
 	// log entires rather than has most up-to-dated committed index
 
 	// For example, Leader 1 send log 80 to follower 2 and follower 3
@@ -335,11 +335,10 @@ func (l *Log) SetCommitIndex(index uint64) error {
 	// follower 2 receive the new committed index and update committed index to 80
 	// leader 1 fail to send the committed index to follower 3
 	// follower 3 promote to leader (server 1 and server 2 will vote, since leader 3
-	// has up-to-dated the entries) 
+	// has up-to-dated the entries)
 	// when new leader 3 send heartbeat with committed index = 0 to follower 2,
 	// follower 2 should reply success and let leader 3 update the committed index to 80
 
-
 	if index < l.commitIndex {
 		return nil
 	}

peer.go

@@ -163,6 +163,11 @@ func (p *Peer) sendFlushRequest(req *AppendEntriesRequest) (uint64, bool, error)
 	// result.
 	//debugln("flush to ", p.Name())
 	debugln("[HeartBeat] Leader ", p.server.Name(), " to ", p.Name(), " ", len(req.Entries), " ", time.Now())
+
+	if p.server.State() != Leader {
+		return 0, false, errors.New("Not leader anymore")
+	}
+
 	resp, err := p.server.transporter.SendAppendEntriesRequest(p.server, p, req)
 
 	//debugln("receive flush response from ", p.Name())
@@ -180,14 +185,24 @@ func (p *Peer) sendFlushRequest(req *AppendEntriesRequest) (uint64, bool, error)
 			debugln("Peer ", p.Name(), "'s' log update to ", p.prevLogIndex)
 		}
 	} else {
+
+		if p.server.State() != Leader {
+			return 0, false, errors.New("Not leader anymore")
+		}
+
+		if resp.Term > p.server.currentTerm {
+			return resp.Term, false, errors.New("Step down")
+		}
 		// Decrement the previous log index down until we find a match. Don't
 		// let it go below where the peer's commit index is though. That's a
 		// problem.
 		if p.prevLogIndex > 0 {
 			p.prevLogIndex--
 		}
 		if resp.CommitIndex > p.prevLogIndex {
-			p.prevLogIndex = resp.CommitIndex
+			debugln("%v %v %v %v", resp.CommitIndex, p.prevLogIndex,
+				p.server.currentTerm, resp.Term)
+			panic("commitedIndex is greater than prevLogIndex")
 		}
 	}
 

server.go

@@ -175,6 +175,13 @@ func (s *Server) Term() uint64 {
 	return s.currentTerm
 }
 
+// Retrieves the current committed index of the server.
+func (s *Server) CommittedIndex() uint64 {
+
+	return s.log.CommitIndex()
+
+}
+
 // Retrieves the name of the candidate this server voted for in this term.
 func (s *Server) VotedFor() string {
 	return s.votedFor

server_test.go

@@ -146,7 +146,6 @@ func TestServerPromote(t *testing.T) {
 
 	leader := servers[0]
 
-
 	if success, err := leader.promote(); !(success && err == nil && leader.state == Leader) {
 		t.Fatalf("Server self-promotion failed: %v (%v)", leader.state, err)
 	}
@@ -182,7 +181,7 @@ func TestServerPromoteDoubleElection(t *testing.T) {
 	if success, err := leader.promote(); !(success && err == nil && leader.state == Leader) {
 		t.Fatalf("Server self-promotion failed: %v (%v)", leader.state, err)
 	}
-	
+
 	if lookup["2"].votedFor != "1" {
 		t.Fatalf("Unexpected vote for server 2: %v", lookup["2"].votedFor)
 	}