fix: prevLogIndex error

Author: treeforest

config.go

@@ -54,8 +54,8 @@ func DefaultConfig() *Config {
 		Address:                 "localhost:4399",
 		MaxLogEntriesPerRequest: 40,
 		HeartbeatInterval:       time.Millisecond * 100,
-		HeartbeatTimeout:        time.Millisecond * 300,
-		ElectionTimeout:         time.Millisecond * 200,
+		HeartbeatTimeout:        time.Millisecond * 250,
+		ElectionTimeout:         time.Millisecond * 300,
 		DialTimeout:             time.Millisecond * 300,
 		DialOptions: []grpc.DialOption{
 			grpc.WithTransportCredentials(insecure.NewCredentials()),

example/kv/main.go

@@ -10,7 +10,6 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"strings"
 	"sync"
 	"time"
 )
@@ -161,7 +160,7 @@ func main() {
 	}
 
 	if *existing != "" {
-		peer.Join(strings.Split(*existing, ","))
+		peer.Join(*existing)
 	}
 
 	done := make(chan os.Signal, 1)

log.go

@@ -219,7 +219,7 @@ func (l *Log) getEntriesAfter(index uint64, maxLogEntriesPerRequest uint64) ([]p
 
 	// Return an error if the index doesn't exist.
 	if index > (uint64(len(l.entries)) + l.startIndex) {
-		panic(fmt.Sprintf("raft: Index is beyond end of log: %v %v", len(l.entries), index))
+		panic(fmt.Sprintf("index is beyond end of log: %v %v", len(l.entries), index))
 	}
 
 	// If we're going from the beginning of the log then return the whole log.
@@ -325,38 +325,37 @@ func (l *Log) flushCommitIndex() {
 	_ = l.levelDB.Put([]byte(commitIndexKey), uint64ToBytes(l.commitIndex), nil)
 }
 
-// Truncates the log to the given index and term. This only works if the log
-// at the index has not been committed.
+// truncate 截断index之后的未提交的所有日志条目，如果已提交或超出日志范围，则返回错误
 func (l *Log) truncate(index uint64, term uint64) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
 	log.Debug("log.truncate: ", index)
 
-	// Do not allow committed entries to be truncated.
+	// 不允许截断已经提交的日志，即要求 index >= commitIndex
 	if index < l.commitIndex {
 		log.Debug("log.truncate.before")
-		return fmt.Errorf("raft.Log: Index is already committed (%v): (IDX=%v, TERM=%v)", l.commitIndex, index, term)
+		return fmt.Errorf("index is already committed (%v): (IDX=%v, TERM=%v)", l.commitIndex, index, term)
 	}
 
-	// Do not truncate past end of entries.
+	// 要截断的日志超出当前日志范围，返回错误
 	if index > l.startIndex+uint64(len(l.entries)) {
 		log.Debug("log.truncate.after")
-		return fmt.Errorf("raft.Log: Entry index does not exist (MAX=%v): (IDX=%v, TERM=%v)", len(l.entries), index, term)
+		return fmt.Errorf("entry index does not exist (MAX=%v): (IDX=%v, TERM=%v)", len(l.entries), index, term)
 	}
 
-	// If we're truncating everything then just clear the entries.
+	// 开始截断操作
 	if index == l.startIndex {
 		log.Debug("log.truncate.clear")
 		l.entries = []pb.LogEntry{}
 	} else {
-		// Do not truncate if the Entry at index does not have the matching term.
+		// 相同索引，却是不同任期，则不进行截断，并返回错误
 		entry := l.entries[index-l.startIndex-1]
 		if len(l.entries) > 0 && entry.Term != term {
 			log.Debug("log.truncate.termMismatch")
 			return fmt.Errorf("entry at index does not have matching term (%v): (IDX=%v, TERM=%v)", entry.Term, index, term)
 		}
 
-		// otherwise, truncate up to the desired Entry.
+		// 截断所要求的条目
 		if index < l.startIndex+uint64(len(l.entries)) {
 			log.Debug("log.truncate.finish")
 			for i := index - l.startIndex; i < uint64(len(l.entries)); i++ {

server.go

@@ -234,7 +234,6 @@ func (s *server) AppendEntries(ctx context.Context, req *pb.AppendEntriesRequest
 		log.Warn("stale term")
 		return resp, nil
 	} else if req.Term == currentTerm {
-		// TODO: 相同任期，都是leader的情况
 		if s.State() == pb.NodeState_Candidate {
 			s.setState(pb.NodeState_Follower)
 		}
@@ -251,7 +250,7 @@ func (s *server) AppendEntries(ctx context.Context, req *pb.AppendEntriesRequest
 	// 2、reply false if log doesn't contain an Entry at prevLogIndex whose
 	// term matches prevLogTerm
 	if err = s.log.truncate(req.PrevLogIndex, req.PrevLogTerm); err != nil {
-		log.Warn(err)
+		log.Debug(err)
 		return resp, nil
 	}
 
@@ -341,6 +340,7 @@ func (s *server) Membership(ctx context.Context, req *pb.MembershipRequest) (
 			continue
 		}
 		s.members.Store(m.Id, newMember(m, s))
+		log.Infof("add member %d %s", m.Id, m.Address)
 	}
 	return &pb.MembershipResponse{Success: true}, nil
 }
@@ -383,6 +383,8 @@ func (s *server) Stop() {
 		return
 	}
 
+	s.leaveCluster()
+
 	// 结束运行的事件循环
 	close(s.stopped)
 
@@ -394,7 +396,31 @@ func (s *server) Stop() {
 	s.log.close()
 	s.setState(pb.NodeState_Stopped)
 
-	log.Info("state")
+	log.Info("stopped")
+}
+
+// leaveCluster 离开集群
+func (s *server) leaveCluster() {
+	if s.IsLeader() {
+		s.members.Range(func(key, value interface{}) bool {
+			m := value.(*member)
+			_ = m.sendRemoveMember(&pb.MemberRequest{Leader: true, Member: s.Self()})
+			return true
+		})
+		return
+	}
+
+	if s.State() == pb.NodeState_Candidate {
+		return
+	}
+
+	value, ok := s.members.Load(s.leaderId)
+	if !ok {
+		return
+	}
+
+	leader := value.(*member)
+	_ = leader.sendRemoveMember(&pb.MemberRequest{Leader: false, Member: s.Self()})
 }
 
 func (s *server) IsLeader() bool {
@@ -649,6 +675,7 @@ func (s *server) leaderLoop() {
 
 		case resp := <-s.leaderRespChan:
 			if resp.Term > s.CurrentTerm() {
+				// 主动退位
 				s.updateCurrentTerm(resp.Term, 0)
 				break
 			}
@@ -703,6 +730,7 @@ func (s *server) addMember(mem pb.Member) {
 	m := newMember(mem, s)
 
 	if s.State() == pb.NodeState_Leader {
+		m.setPrevLogIndex(s.log.CurrentIndex())
 		m.startHeartbeat()
 	}
 
@@ -741,6 +769,7 @@ func (s *server) removeMember(id uint64) {
 	if !loaded {
 		return
 	}
+	log.Infof("remove member %d", id)
 
 	if s.IsLeader() {
 		m := val.(*member)