Merge pull request #1203 from coreos/fix_raft

raft: commitIndex=min(leaderCommit, index of last new entry)

raft/log.go

@@ -47,6 +47,7 @@ func (l *raftLog) String() string {
 }
 
 func (l *raftLog) maybeAppend(index, logTerm, committed int64, ents ...pb.Entry) bool {
+	lastnewi := index + int64(len(ents))
 	if l.matchTerm(index, logTerm) {
 		from := index + 1
 		ci := l.findConflict(from, ents)
@@ -57,8 +58,10 @@ func (l *raftLog) maybeAppend(index, logTerm, committed int64, ents ...pb.Entry)
 		default:
 			l.append(ci-1, ents[ci-from:]...)
 		}
-		if l.committed < committed {
-			l.committed = min(committed, l.lastIndex())
+		tocommit := min(committed, lastnewi)
+		// if toCommit > commitIndex, set commitIndex = toCommit
+		if l.committed < tocommit {
+			l.committed = tocommit
 		}
 		return true
 	}

raft/raft_test.go

@@ -489,8 +489,10 @@ func TestHandleMsgApp(t *testing.T) {
 		{pb.Message{Type: msgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 4, Entries: []pb.Entry{{Term: 2}}}, 2, 2, false},
 
 		// Ensure 3
-		{pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 2}, 2, 2, false},
-		{pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4}, 2, 2, false}, // commit upto min(commit, last)
+		{pb.Message{Type: msgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3}, 2, 1, false},                                 // match entry 1, commit upto last new entry 1
+		{pb.Message{Type: msgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3, Entries: []pb.Entry{{Term: 2}}}, 2, 2, false}, // match entry 1, commit upto last new entry 2
+		{pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3}, 2, 2, false},                                 // match entry 2, commit upto last new entry 2
+		{pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4}, 2, 2, false},                                 // commit upto log.last()
 	}
 
 	for i, tt := range tests {