diff --git a/fuzzy/cluster.go b/fuzzy/cluster.go index df4293e8..33554319 100644 --- a/fuzzy/cluster.go +++ b/fuzzy/cluster.go @@ -5,6 +5,7 @@ package fuzzy import ( "bytes" + "errors" "fmt" "io" "os" @@ -22,6 +23,14 @@ type appliedItem struct { data []byte } +type staticErrorFuture struct { + err error +} + +func (s staticErrorFuture) Error() error { + return s.err +} + type cluster struct { nodes []*raftNode removedNodes []*raftNode @@ -230,6 +239,9 @@ func (c *cluster) generateNApplies(s *applySource, n uint) [][]byte { func (c *cluster) leadershipTransfer(leaderTimeout time.Duration) raft.Future { ldr := c.Leader(leaderTimeout) + if ldr == nil { + return staticErrorFuture{err: errors.New("no leader elected before leadership transfer")} + } return ldr.raft.LeadershipTransfer() } diff --git a/fuzzy/leadershiptransfer_test.go b/fuzzy/leadershiptransfer_test.go index 781107f7..9be8a9d7 100644 --- a/fuzzy/leadershiptransfer_test.go +++ b/fuzzy/leadershiptransfer_test.go @@ -19,15 +19,21 @@ func TestRaft_FuzzyLeadershipTransfer(t *testing.T) { s := newApplySource("LeadershipTransfer") data := cluster.generateNApplies(s, uint(r.Intn(10000))) futures := cluster.sendNApplies(time.Minute, data) - cluster.leadershipTransfer(time.Minute) + if err := cluster.leadershipTransfer(time.Minute).Error(); err != nil { + t.Fatalf("leadership transfer failed: %v", err) + } data = cluster.generateNApplies(s, uint(r.Intn(10000))) futures = append(futures, cluster.sendNApplies(time.Minute, data)...) - cluster.leadershipTransfer(time.Minute) + if err := cluster.leadershipTransfer(time.Minute).Error(); err != nil { + t.Fatalf("leadership transfer failed: %v", err) + } data = cluster.generateNApplies(s, uint(r.Intn(10000))) futures = append(futures, cluster.sendNApplies(time.Minute, data)...) - cluster.leadershipTransfer(time.Minute) + if err := cluster.leadershipTransfer(time.Minute).Error(); err != nil { + t.Fatalf("leadership transfer failed: %v", err) + } data = cluster.generateNApplies(s, uint(r.Intn(10000))) futures = append(futures, cluster.sendNApplies(time.Minute, data)...) @@ -39,6 +45,20 @@ func TestRaft_FuzzyLeadershipTransfer(t *testing.T) { cluster.VerifyFSM(t) } +func TestRaft_FuzzyLeadershipTransferWithoutLeader(t *testing.T) { + cluster := newRaftCluster(t, testLogWriter, "lt_no_leader", 5, nil) + + if err := cluster.leadershipTransfer(0).Error(); err == nil { + t.Fatal("expected leadership transfer to fail when no leader can be elected") + } + + for _, n := range cluster.nodes { + if err := n.raft.Shutdown().Error(); err != nil { + t.Fatalf("shutdown failed for %s: %v", n.name, err) + } + } +} + type LeadershipTransferMode int type LeadershipTransfer struct { diff --git a/fuzzy/transport.go b/fuzzy/transport.go index 2bcfd8a6..9fa5c690 100644 --- a/fuzzy/transport.go +++ b/fuzzy/transport.go @@ -113,6 +113,7 @@ func (t *transport) sendRPC(target string, req interface{}, resp interface{}) er } rpc := raft.RPC{RespChan: rc} var reqVote raft.RequestVoteRequest + var reqPreVote raft.RequestPreVoteRequest var timeoutNow raft.TimeoutNowRequest var appEnt raft.AppendEntriesRequest dec := codec.NewDecoderBytes(buff.Bytes(), &codecHandle) @@ -127,6 +128,11 @@ func (t *transport) sendRPC(target string, req interface{}, resp interface{}) er return err } rpc.Command = &reqVote + case *raft.RequestPreVoteRequest: + if err := dec.Decode(&reqPreVote); err != nil { + return err + } + rpc.Command = &reqPreVote case *raft.AppendEntriesRequest: if err := dec.Decode(&appEnt); err != nil { return err