| // Copyright 2016 PingCAP, Inc. |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| // Copyright 2015 CoreOS, Inc. |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| use protobuf::RepeatedField; |
| use raft::eraftpb::*; |
| use raft::storage::MemStorage; |
| use raft::*; |
| use test_util::*; |
| |
| fn commit_noop_entry(r: &mut Interface, s: &MemStorage) { |
| assert_eq!(r.state, StateRole::Leader); |
| r.bcast_append(); |
| // simulate the response of MsgAppend |
| let msgs = r.read_messages(); |
| for m in msgs { |
| assert_eq!(m.get_msg_type(), MessageType::MsgAppend); |
| assert_eq!(m.get_entries().len(), 1); |
| assert!(m.get_entries()[0].get_data().is_empty()); |
| r.step(accept_and_reply(&m)).expect(""); |
| } |
| // ignore further messages to refresh followers' commit index |
| r.read_messages(); |
| s.wl() |
| .append(r.raft_log.unstable_entries().unwrap_or(&[])) |
| .expect(""); |
| let committed = r.raft_log.committed; |
| r.raft_log.applied_to(committed); |
| let (last_index, last_term) = (r.raft_log.last_index(), r.raft_log.last_term()); |
| r.raft_log.stable_to(last_index, last_term); |
| } |
| |
| fn accept_and_reply(m: &Message) -> Message { |
| assert_eq!(m.get_msg_type(), MessageType::MsgAppend); |
| let mut reply = new_message(m.get_to(), m.get_from(), MessageType::MsgAppendResponse, 0); |
| reply.set_term(m.get_term()); |
| reply.set_index(m.get_index() + m.get_entries().len() as u64); |
| reply |
| } |
| |
| #[test] |
| fn test_follower_update_term_from_message() { |
| setup_for_test(); |
| test_update_term_from_message(StateRole::Follower); |
| } |
| |
| #[test] |
| fn test_candidate_update_term_from_message() { |
| setup_for_test(); |
| test_update_term_from_message(StateRole::Candidate); |
| } |
| |
| #[test] |
| fn test_leader_update_term_from_message() { |
| setup_for_test(); |
| test_update_term_from_message(StateRole::Leader); |
| } |
| |
| // test_update_term_from_message tests that if one server’s current term is |
| // smaller than the other’s, then it updates its current term to the larger |
| // value. If a candidate or leader discovers that its term is out of date, |
| // it immediately reverts to follower state. |
| // Reference: section 5.1 |
| fn test_update_term_from_message(state: StateRole) { |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| match state { |
| StateRole::Follower => r.become_follower(1, 2), |
| StateRole::PreCandidate => r.become_pre_candidate(), |
| StateRole::Candidate => r.become_candidate(), |
| StateRole::Leader => { |
| r.become_candidate(); |
| r.become_leader(); |
| } |
| } |
| |
| let mut m = new_message(0, 0, MessageType::MsgAppend, 0); |
| m.set_term(2); |
| r.step(m).expect(""); |
| |
| assert_eq!(r.term, 2); |
| assert_eq!(r.state, StateRole::Follower); |
| } |
| |
| // test_start_as_follower tests that when servers start up, they begin as followers. |
| // Reference: section 5.2 |
| #[test] |
| fn test_start_as_follower() { |
| setup_for_test(); |
| let r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| assert_eq!(r.state, StateRole::Follower); |
| } |
| |
| // test_leader_bcast_beat tests that if the leader receives a heartbeat tick, |
| // it will send a msgApp with m.Index = 0, m.LogTerm=0 and empty entries as |
| // heartbeat to all followers. |
| // Reference: section 5.2 |
| #[test] |
| fn test_leader_bcast_beat() { |
| setup_for_test(); |
| // heartbeat interval |
| let hi = 1; |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, hi, new_storage()); |
| r.become_candidate(); |
| r.become_leader(); |
| for i in 0..10 { |
| r.append_entry(&mut [empty_entry(0, i as u64 + 1)]); |
| } |
| |
| for _ in 0..hi { |
| r.tick(); |
| } |
| |
| let mut msgs = r.read_messages(); |
| msgs.sort_by_key(|m| format!("{:?}", m)); |
| |
| let new_message_ext = |f, to| { |
| let mut m = new_message(f, to, MessageType::MsgHeartbeat, 0); |
| m.set_term(1); |
| m.set_commit(0); |
| m |
| }; |
| |
| let expect_msgs = vec![new_message_ext(1, 2), new_message_ext(1, 3)]; |
| assert_eq!(msgs, expect_msgs); |
| } |
| |
| #[test] |
| fn test_follower_start_election() { |
| setup_for_test(); |
| test_nonleader_start_election(StateRole::Follower); |
| } |
| |
| #[test] |
| fn test_candidate_start_new_election() { |
| setup_for_test(); |
| test_nonleader_start_election(StateRole::Candidate); |
| } |
| |
| // test_nonleader_start_election tests that if a follower receives no communication |
| // over election timeout, it begins an election to choose a new leader. It |
| // increments its current term and transitions to candidate state. It then |
| // votes for itself and issues RequestVote RPCs in parallel to each of the |
| // other servers in the cluster. |
| // Reference: section 5.2 |
| // Also if a candidate fails to obtain a majority, it will time out and |
| // start a new election by incrementing its term and initiating another |
| // round of RequestVote RPCs. |
| // Reference: section 5.2 |
| fn test_nonleader_start_election(state: StateRole) { |
| // election timeout |
| let et = 10; |
| let mut r = new_test_raft(1, vec![1, 2, 3], et, 1, new_storage()); |
| match state { |
| StateRole::Follower => r.become_follower(1, 2), |
| StateRole::Candidate => r.become_candidate(), |
| _ => panic!("Only non-leader role is accepted."), |
| } |
| |
| for _ in 1..2 * et { |
| r.tick(); |
| } |
| |
| assert_eq!(r.term, 2); |
| assert_eq!(r.state, StateRole::Candidate); |
| assert!(r.votes[&r.id]); |
| let mut msgs = r.read_messages(); |
| msgs.sort_by_key(|m| format!("{:?}", m)); |
| let new_message_ext = |f, to| { |
| let mut m = new_message(f, to, MessageType::MsgRequestVote, 0); |
| m.set_term(2); |
| m.set_log_term(0); |
| m.set_index(0); |
| m |
| }; |
| let expect_msgs = vec![new_message_ext(1, 2), new_message_ext(1, 3)]; |
| assert_eq!(msgs, expect_msgs); |
| } |
| |
| // test_leader_election_in_one_round_rpc tests all cases that may happen in |
| // leader election during one round of RequestVote RPC: |
| // a) it wins the election |
| // b) it loses the election |
| // c) it is unclear about the result |
| // Reference: section 5.2 |
| #[test] |
| fn test_leader_election_in_one_round_rpc() { |
| setup_for_test(); |
| let mut tests = vec![ |
| // win the election when receiving votes from a majority of the servers |
| (1, map!(), StateRole::Leader), |
| (3, map!(2 => true, 3 => true), StateRole::Leader), |
| (3, map!(2 => true), StateRole::Leader), |
| ( |
| 5, |
| map!(2 => true, 3 => true, 4 => true, 5 => true), |
| StateRole::Leader, |
| ), |
| (5, map!(2 => true, 3 => true, 4 => true), StateRole::Leader), |
| (5, map!(2 => true, 3 => true), StateRole::Leader), |
| // return to follower state if it receives vote denial from a majority |
| (3, map!(2 => false, 3 => false), StateRole::Follower), |
| ( |
| 5, |
| map!(2 => false, 3 => false, 4 => false, 5 => false), |
| StateRole::Follower, |
| ), |
| ( |
| 5, |
| map!(2 => true, 3 => false, 4 => false, 5 => false), |
| StateRole::Follower, |
| ), |
| // stay in candidate if it does not obtain the majority |
| (3, map!(), StateRole::Candidate), |
| (5, map!(2 => true), StateRole::Candidate), |
| (5, map!(2 => false, 3 => false), StateRole::Candidate), |
| (5, map!(), StateRole::Candidate), |
| ]; |
| |
| for (i, (size, votes, state)) in tests.drain(..).enumerate() { |
| let mut r = new_test_raft(1, (1..size as u64 + 1).collect(), 10, 1, new_storage()); |
| |
| r.step(new_message(1, 1, MessageType::MsgHup, 0)).expect(""); |
| for (id, vote) in votes { |
| let mut m = new_message(id, 1, MessageType::MsgRequestVoteResponse, 0); |
| m.set_term(r.term); |
| m.set_reject(!vote); |
| r.step(m).expect(""); |
| } |
| |
| if r.state != state { |
| panic!("#{}: state = {:?}, want {:?}", i, r.state, state); |
| } |
| if r.term != 1 { |
| panic!("#{}: term = {}, want {}", i, r.term, 1); |
| } |
| } |
| } |
| |
| // test_follower_vote tests that each follower will vote for at most one |
| // candidate in a given term, on a first-come-first-served basis. |
| // Reference: section 5.2 |
| #[test] |
| fn test_follower_vote() { |
| setup_for_test(); |
| let mut tests = vec![ |
| (INVALID_ID, 1, false), |
| (INVALID_ID, 2, false), |
| (1, 1, false), |
| (2, 2, false), |
| (1, 2, true), |
| (2, 1, true), |
| ]; |
| |
| for (i, (vote, nvote, wreject)) in tests.drain(..).enumerate() { |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| r.load_state(&hard_state(1, 0, vote)); |
| |
| let mut m = new_message(nvote, 1, MessageType::MsgRequestVote, 0); |
| m.set_term(1); |
| r.step(m).expect(""); |
| |
| let msgs = r.read_messages(); |
| let mut m = new_message(1, nvote, MessageType::MsgRequestVoteResponse, 0); |
| m.set_term(1); |
| m.set_reject(wreject); |
| let expect_msgs = vec![m]; |
| if msgs != expect_msgs { |
| panic!("#{}: msgs = {:?}, want {:?}", i, msgs, expect_msgs); |
| } |
| } |
| } |
| |
| // test_candidate_fallback tests that while waiting for votes, |
| // if a candidate receives an AppendEntries RPC from another server claiming |
| // to be leader whose term is at least as large as the candidate's current term, |
| // it recognizes the leader as legitimate and returns to follower state. |
| // Reference: section 5.2 |
| #[test] |
| fn test_candidate_fallback() { |
| setup_for_test(); |
| let new_message_ext = |f, to, term| { |
| let mut m = new_message(f, to, MessageType::MsgAppend, 0); |
| m.set_term(term); |
| m |
| }; |
| let mut tests = vec![new_message_ext(2, 1, 1), new_message_ext(2, 1, 2)]; |
| for (i, m) in tests.drain(..).enumerate() { |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| r.step(new_message(1, 1, MessageType::MsgHup, 0)).expect(""); |
| assert_eq!(r.state, StateRole::Candidate); |
| |
| let term = m.get_term(); |
| r.step(m).expect(""); |
| |
| if r.state != StateRole::Follower { |
| panic!( |
| "#{}: state = {:?}, want {:?}", |
| i, |
| r.state, |
| StateRole::Follower |
| ); |
| } |
| if r.term != term { |
| panic!("#{}: term = {}, want {}", i, r.term, term); |
| } |
| } |
| } |
| |
| #[test] |
| fn test_follower_election_timeout_randomized() { |
| setup_for_test(); |
| test_non_leader_election_timeout_randomized(StateRole::Follower); |
| } |
| |
| #[test] |
| fn test_candidate_election_timeout_randomized() { |
| setup_for_test(); |
| test_non_leader_election_timeout_randomized(StateRole::Candidate); |
| } |
| |
| // test_non_leader_election_timeout_randomized tests that election timeout for |
| // follower or candidate is randomized. |
| // Reference: section 5.2 |
| fn test_non_leader_election_timeout_randomized(state: StateRole) { |
| let et = 10; |
| let mut r = new_test_raft(1, vec![1, 2, 3], et, 1, new_storage()); |
| let mut timeouts = map!(); |
| for _ in 0..1000 * et { |
| let term = r.term; |
| match state { |
| StateRole::Follower => r.become_follower(term + 1, 2), |
| StateRole::Candidate => r.become_candidate(), |
| _ => panic!("only non leader state is accepted!"), |
| } |
| |
| let mut time = 0; |
| while r.read_messages().is_empty() { |
| r.tick(); |
| time += 1; |
| } |
| timeouts.insert(time, true); |
| } |
| |
| assert!(timeouts.len() <= et && timeouts.len() >= et - 1); |
| for d in et + 1..2 * et { |
| assert!(timeouts[&d]); |
| } |
| } |
| |
| #[test] |
| fn test_follower_election_timeout_nonconflict() { |
| setup_for_test(); |
| test_nonleaders_election_timeout_nonconfict(StateRole::Follower); |
| } |
| |
| #[test] |
| fn test_acandidates_election_timeout_nonconf() { |
| setup_for_test(); |
| test_nonleaders_election_timeout_nonconfict(StateRole::Candidate); |
| } |
| |
| // test_nonleaders_election_timeout_nonconfict tests that in most cases only a |
| // single server(follower or candidate) will time out, which reduces the |
| // likelihood of split vote in the new election. |
| // Reference: section 5.2 |
| fn test_nonleaders_election_timeout_nonconfict(state: StateRole) { |
| let et = 10; |
| let size = 5; |
| let mut rs = Vec::with_capacity(size); |
| let ids: Vec<u64> = (1..size as u64 + 1).collect(); |
| for id in ids.iter().take(size) { |
| rs.push(new_test_raft(*id, ids.clone(), et, 1, new_storage())); |
| } |
| let mut conflicts = 0; |
| for _ in 0..1000 { |
| for r in &mut rs { |
| let term = r.term; |
| match state { |
| StateRole::Follower => r.become_follower(term + 1, INVALID_ID), |
| StateRole::Candidate => r.become_candidate(), |
| _ => panic!("non leader state is expect!"), |
| } |
| } |
| |
| let mut timeout_num = 0; |
| while timeout_num == 0 { |
| for r in &mut rs { |
| r.tick(); |
| if !r.read_messages().is_empty() { |
| timeout_num += 1; |
| } |
| } |
| } |
| // several rafts time out at the same tick |
| if timeout_num > 1 { |
| conflicts += 1; |
| } |
| } |
| |
| assert!(f64::from(conflicts) / 1000.0 <= 0.3); |
| } |
| |
| // test_leader_start_replication tests that when receiving client proposals, |
| // the leader appends the proposal to its log as a new entry, then issues |
| // AppendEntries RPCs in parallel to each of the other servers to replicate |
| // the entry. Also, when sending an AppendEntries RPC, the leader includes |
| // the index and term of the entry in its log that immediately precedes |
| // the new entries. |
| // Also, it writes the new entry into stable storage. |
| // Reference: section 5.3 |
| #[test] |
| fn test_leader_start_replication() { |
| setup_for_test(); |
| let s = new_storage(); |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, s.clone()); |
| r.become_candidate(); |
| r.become_leader(); |
| commit_noop_entry(&mut r, &s); |
| let li = r.raft_log.last_index(); |
| |
| r.step(new_message(1, 1, MessageType::MsgPropose, 1)) |
| .expect(""); |
| |
| assert_eq!(r.raft_log.last_index(), li + 1); |
| assert_eq!(r.raft_log.committed, li); |
| let mut msgs = r.read_messages(); |
| msgs.sort_by_key(|m| format!("{:?}", m)); |
| let wents = vec![new_entry(1, li + 1, SOME_DATA)]; |
| let new_message_ext = |f, to, ents| { |
| let mut m = new_message(f, to, MessageType::MsgAppend, 0); |
| m.set_term(1); |
| m.set_index(li); |
| m.set_log_term(1); |
| m.set_commit(li); |
| m.set_entries(RepeatedField::from_vec(ents)); |
| m |
| }; |
| let expect_msgs = vec![ |
| new_message_ext(1, 2, wents.clone()), |
| new_message_ext(1, 3, wents.clone()), |
| ]; |
| assert_eq!(msgs, expect_msgs); |
| assert_eq!(r.raft_log.unstable_entries(), Some(&*wents)); |
| } |
| |
| // test_leader_commit_entry tests that when the entry has been safely replicated, |
| // the leader gives out the applied entries, which can be applied to its state |
| // machine. |
| // Also, the leader keeps track of the highest index it knows to be committed, |
| // and it includes that index in future AppendEntries RPCs so that the other |
| // servers eventually find out. |
| // Reference: section 5.3 |
| #[test] |
| fn test_leader_commit_entry() { |
| setup_for_test(); |
| let s = new_storage(); |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, s.clone()); |
| r.become_candidate(); |
| r.become_leader(); |
| commit_noop_entry(&mut r, &s); |
| let li = r.raft_log.last_index(); |
| r.step(new_message(1, 1, MessageType::MsgPropose, 1)) |
| .expect(""); |
| |
| for m in r.read_messages() { |
| r.step(accept_and_reply(&m)).expect(""); |
| } |
| |
| assert_eq!(r.raft_log.committed, li + 1); |
| let wents = vec![new_entry(1, li + 1, SOME_DATA)]; |
| assert_eq!(r.raft_log.next_entries(), Some(wents)); |
| let mut msgs = r.read_messages(); |
| msgs.sort_by_key(|m| format!("{:?}", m)); |
| for (i, m) in msgs.drain(..).enumerate() { |
| assert_eq!(i as u64 + 2, m.get_to()); |
| assert_eq!(m.get_msg_type(), MessageType::MsgAppend); |
| assert_eq!(m.get_commit(), li + 1); |
| } |
| } |
| |
| // test_leader_acknowledge_commit tests that a log entry is committed once the |
| // leader that created the entry has replicated it on a majority of the servers. |
| // Reference: section 5.3 |
| #[test] |
| fn test_leader_acknowledge_commit() { |
| setup_for_test(); |
| let mut tests = vec![ |
| (1, map!(), true), |
| (3, map!(), false), |
| (3, map!(2 => true), true), |
| (3, map!(2 => true, 3 => true), true), |
| (5, map!(), false), |
| (5, map!(2 => true), false), |
| (5, map!(2 => true, 3 => true), true), |
| (5, map!(2 => true, 3 => true, 4 => true), true), |
| (5, map!(2 => true, 3 => true, 4 => true, 5 => true), true), |
| ]; |
| for (i, (size, acceptors, wack)) in tests.drain(..).enumerate() { |
| let s = new_storage(); |
| let mut r = new_test_raft(1, (1..size + 1).collect(), 10, 1, s.clone()); |
| r.become_candidate(); |
| r.become_leader(); |
| commit_noop_entry(&mut r, &s); |
| let li = r.raft_log.last_index(); |
| r.step(new_message(1, 1, MessageType::MsgPropose, 1)) |
| .expect(""); |
| |
| for m in r.read_messages() { |
| if acceptors.contains_key(&m.get_to()) && acceptors[&m.get_to()] { |
| r.step(accept_and_reply(&m)).expect(""); |
| } |
| } |
| |
| let g = r.raft_log.committed > li; |
| if g ^ wack { |
| panic!("#{}: ack commit = {}, want {}", i, g, wack); |
| } |
| } |
| } |
| |
| // test_leader_commit_preceding_entries tests that when leader commits a log entry, |
| // it also commits all preceding entries in the leader’s log, including |
| // entries created by previous leaders. |
| // Also, it applies the entry to its local state machine (in log order). |
| // Reference: section 5.3 |
| #[test] |
| fn test_leader_commit_preceding_entries() { |
| setup_for_test(); |
| let mut tests = vec![ |
| vec![], |
| vec![empty_entry(2, 1)], |
| vec![empty_entry(1, 1), empty_entry(2, 2)], |
| vec![empty_entry(1, 1)], |
| ]; |
| |
| for (i, mut tt) in tests.drain(..).enumerate() { |
| let s = new_storage(); |
| s.wl().append(&tt).expect(""); |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, s); |
| r.load_state(&hard_state(2, 0, 0)); |
| r.become_candidate(); |
| r.become_leader(); |
| r.step(new_message(1, 1, MessageType::MsgPropose, 1)) |
| .expect(""); |
| |
| for m in r.read_messages() { |
| r.step(accept_and_reply(&m)).expect(""); |
| } |
| |
| let li = tt.len() as u64; |
| tt.append(&mut vec![ |
| empty_entry(3, li + 1), |
| new_entry(3, li + 2, SOME_DATA), |
| ]); |
| let g = r.raft_log.next_entries(); |
| let wg = Some(tt); |
| if g != wg { |
| panic!("#{}: ents = {:?}, want {:?}", i, g, wg); |
| } |
| } |
| } |
| |
| // test_follower_commit_entry tests that once a follower learns that a log entry |
| // is committed, it applies the entry to its local state machine (in log order). |
| // Reference: section 5.3 |
| #[test] |
| fn test_follower_commit_entry() { |
| setup_for_test(); |
| let mut tests = vec![ |
| (vec![new_entry(1, 1, SOME_DATA)], 1), |
| ( |
| vec![ |
| new_entry(1, 1, SOME_DATA), |
| new_entry(1, 2, Some("somedata2")), |
| ], |
| 2, |
| ), |
| ( |
| vec![ |
| new_entry(1, 1, Some("somedata2")), |
| new_entry(1, 2, SOME_DATA), |
| ], |
| 2, |
| ), |
| ( |
| vec![ |
| new_entry(1, 1, SOME_DATA), |
| new_entry(1, 2, Some("somedata2")), |
| ], |
| 1, |
| ), |
| ]; |
| |
| for (i, (ents, commit)) in tests.drain(..).enumerate() { |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| r.become_follower(1, 2); |
| |
| let mut m = new_message(2, 1, MessageType::MsgAppend, 0); |
| m.set_term(1); |
| m.set_commit(commit); |
| m.set_entries(RepeatedField::from_vec(ents.clone())); |
| r.step(m).expect(""); |
| |
| if r.raft_log.committed != commit { |
| panic!( |
| "#{}: committed = {}, want {}", |
| i, r.raft_log.committed, commit |
| ); |
| } |
| let wents = Some(ents[..commit as usize].to_vec()); |
| let g = r.raft_log.next_entries(); |
| if g != wents { |
| panic!("#{}: next_ents = {:?}, want {:?}", i, g, wents); |
| } |
| } |
| } |
| |
| // test_follower_check_msg_append tests that if the follower does not find an |
| // entry in its log with the same index and term as the one in AppendEntries RPC, |
| // then it refuses the new entries. Otherwise it replies that it accepts the |
| // append entries. |
| // Reference: section 5.3 |
| #[test] |
| fn test_follower_check_msg_append() { |
| setup_for_test(); |
| let ents = vec![empty_entry(1, 1), empty_entry(2, 2)]; |
| let mut tests = vec![ |
| // match with committed entries |
| (0, 0, 1, false, 0), |
| (ents[0].get_term(), ents[0].get_index(), 1, false, 0), |
| // match with uncommitted entries |
| (ents[1].get_term(), ents[1].get_index(), 2, false, 0), |
| // unmatch with existing entry |
| ( |
| ents[0].get_term(), |
| ents[1].get_index(), |
| ents[1].get_index(), |
| true, |
| 2, |
| ), |
| // unexisting entry |
| ( |
| ents[1].get_term() + 1, |
| ents[1].get_index() + 1, |
| ents[1].get_index() + 1, |
| true, |
| 2, |
| ), |
| ]; |
| for (i, (term, index, windex, wreject, wreject_hint)) in tests.drain(..).enumerate() { |
| let s = new_storage(); |
| s.wl().append(&ents).expect(""); |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, s); |
| r.load_state(&hard_state(0, 1, 0)); |
| r.become_follower(2, 2); |
| |
| let mut m = new_message(2, 1, MessageType::MsgAppend, 0); |
| m.set_term(2); |
| m.set_log_term(term); |
| m.set_index(index); |
| r.step(m).expect(""); |
| |
| let msgs = r.read_messages(); |
| let mut wm = new_message(1, 2, MessageType::MsgAppendResponse, 0); |
| wm.set_term(2); |
| wm.set_index(windex); |
| if wreject { |
| wm.set_reject(wreject); |
| wm.set_reject_hint(wreject_hint); |
| } |
| let expect_msgs = vec![wm]; |
| if msgs != expect_msgs { |
| panic!("#{}: msgs = {:?}, want {:?}", i, msgs, expect_msgs); |
| } |
| } |
| } |
| |
| // test_follower_append_entries tests that when AppendEntries RPC is valid, |
| // the follower will delete the existing conflict entry and all that follow it, |
| // and append any new entries not already in the log. |
| // Also, it writes the new entry into stable storage. |
| // Reference: section 5.3 |
| #[test] |
| fn test_follower_append_entries() { |
| setup_for_test(); |
| let mut tests = vec![ |
| ( |
| 2, |
| 2, |
| vec![empty_entry(3, 3)], |
| vec![empty_entry(1, 1), empty_entry(2, 2), empty_entry(3, 3)], |
| vec![empty_entry(3, 3)], |
| ), |
| ( |
| 1, |
| 1, |
| vec![empty_entry(3, 2), empty_entry(4, 3)], |
| vec![empty_entry(1, 1), empty_entry(3, 2), empty_entry(4, 3)], |
| vec![empty_entry(3, 2), empty_entry(4, 3)], |
| ), |
| ( |
| 0, |
| 0, |
| vec![empty_entry(1, 1)], |
| vec![empty_entry(1, 1), empty_entry(2, 2)], |
| vec![], |
| ), |
| ( |
| 0, |
| 0, |
| vec![empty_entry(3, 1)], |
| vec![empty_entry(3, 1)], |
| vec![empty_entry(3, 1)], |
| ), |
| ]; |
| for (i, (index, term, ents, wents, wunstable)) in tests.drain(..).enumerate() { |
| let s = new_storage(); |
| s.wl() |
| .append(&[empty_entry(1, 1), empty_entry(2, 2)]) |
| .expect(""); |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, s); |
| r.become_follower(2, 2); |
| |
| let mut m = new_message(2, 1, MessageType::MsgAppend, 0); |
| m.set_term(2); |
| m.set_log_term(term); |
| m.set_index(index); |
| m.set_entries(RepeatedField::from_vec(ents)); |
| r.step(m).expect(""); |
| |
| let g = r.raft_log.all_entries(); |
| if g != wents { |
| panic!("#{}: ents = {:?}, want {:?}", i, g, wents); |
| } |
| let g = r.raft_log.unstable_entries(); |
| let wunstable = if wunstable.is_empty() { |
| None |
| } else { |
| Some(&*wunstable) |
| }; |
| if g != wunstable { |
| panic!("#{}: unstable_entries = {:?}, want {:?}", i, g, wunstable); |
| } |
| } |
| } |
| |
| // test_leader_sync_follower_log tests that the leader could bring a follower's log |
| // into consistency with its own. |
| // Reference: section 5.3, figure 7 |
| #[test] |
| fn test_leader_sync_follower_log() { |
| setup_for_test(); |
| let ents = vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| empty_entry(4, 5), |
| empty_entry(5, 6), |
| empty_entry(5, 7), |
| empty_entry(6, 8), |
| empty_entry(6, 9), |
| empty_entry(6, 10), |
| ]; |
| let term = 8u64; |
| let mut tests = vec![ |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| empty_entry(4, 5), |
| empty_entry(5, 6), |
| empty_entry(5, 7), |
| empty_entry(6, 8), |
| empty_entry(6, 9), |
| ], |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| ], |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| empty_entry(4, 5), |
| empty_entry(5, 6), |
| empty_entry(5, 7), |
| empty_entry(6, 8), |
| empty_entry(6, 9), |
| empty_entry(6, 10), |
| empty_entry(6, 11), |
| ], |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| empty_entry(4, 5), |
| empty_entry(5, 6), |
| empty_entry(5, 7), |
| empty_entry(6, 8), |
| empty_entry(6, 9), |
| empty_entry(6, 10), |
| empty_entry(7, 11), |
| empty_entry(7, 12), |
| ], |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(4, 4), |
| empty_entry(4, 5), |
| empty_entry(4, 6), |
| empty_entry(4, 7), |
| ], |
| vec![ |
| empty_entry(0, 0), |
| empty_entry(1, 1), |
| empty_entry(1, 2), |
| empty_entry(1, 3), |
| empty_entry(2, 4), |
| empty_entry(2, 5), |
| empty_entry(2, 6), |
| empty_entry(3, 7), |
| empty_entry(3, 8), |
| empty_entry(3, 9), |
| empty_entry(3, 10), |
| empty_entry(3, 11), |
| ], |
| ]; |
| for (i, tt) in tests.drain(..).enumerate() { |
| let lead_store = new_storage(); |
| lead_store.wl().append(&ents).expect(""); |
| let mut lead = new_test_raft(1, vec![1, 2, 3], 10, 1, lead_store); |
| let last_index = lead.raft_log.last_index(); |
| lead.load_state(&hard_state(term, last_index, 0)); |
| let follower_store = new_storage(); |
| follower_store.wl().append(&tt).expect(""); |
| let mut follower = new_test_raft(2, vec![1, 2, 3], 10, 1, follower_store); |
| follower.load_state(&hard_state(term - 1, 0, 0)); |
| // It is necessary to have a three-node cluster. |
| // The second may have more up-to-date log than the first one, so the |
| // first node needs the vote from the third node to become the leader. |
| let mut n = Network::new(vec![Some(lead), Some(follower), NOP_STEPPER]); |
| n.send(vec![new_message(1, 1, MessageType::MsgHup, 0)]); |
| // The election occurs in the term after the one we loaded with |
| // lead.load_state above. |
| let mut m = new_message(3, 1, MessageType::MsgRequestVoteResponse, 0); |
| m.set_term(term + 1); |
| n.send(vec![m]); |
| |
| let mut m = new_message(1, 1, MessageType::MsgPropose, 0); |
| m.set_entries(RepeatedField::from_vec(vec![Entry::new()])); |
| n.send(vec![m]); |
| let lead_str = ltoa(&n.peers[&1].raft_log); |
| let follower_str = ltoa(&n.peers[&2].raft_log); |
| if lead_str != follower_str { |
| panic!( |
| "#{}: lead str: {}, follower_str: {}", |
| i, lead_str, follower_str |
| ); |
| } |
| } |
| } |
| |
| // test_vote_request tests that the vote request includes information about the candidate’s log |
| // and are sent to all of the other nodes. |
| // Reference: section 5.4.1 |
| #[test] |
| fn test_vote_request() { |
| setup_for_test(); |
| let mut tests = vec![ |
| (vec![empty_entry(1, 1)], 2), |
| (vec![empty_entry(1, 1), empty_entry(2, 2)], 3), |
| ]; |
| for (j, (ents, wterm)) in tests.drain(..).enumerate() { |
| let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage()); |
| let mut m = new_message(2, 1, MessageType::MsgAppend, 0); |
| m.set_term(wterm - 1); |
| m.set_log_term(0); |
| m.set_index(0); |
| m.set_entries(RepeatedField::from_vec(ents.clone())); |
| r.step(m).expect(""); |
| r.read_messages(); |
| |
| for _ in 1..r.get_election_timeout() * 2 { |
| r.tick_election(); |
| } |
| |
| let mut msgs = r.read_messages(); |
| msgs.sort_by_key(|m| format!("{:?}", m)); |
| if msgs.len() != 2 { |
| panic!("#{}: msg count = {}, want 2", j, msgs.len()); |
| } |
| for (i, m) in msgs.iter().enumerate() { |
| if m.get_msg_type() != MessageType::MsgRequestVote { |
| panic!( |
| "#{}.{}: msg_type = {:?}, want {:?}", |
| j, |
| i, |
| m.get_msg_type(), |
| MessageType::MsgRequestVote |
| ); |
| } |
| if m.get_to() != i as u64 + 2 { |
| panic!("#{}.{}: to = {}, want {}", j, i, m.get_to(), i + 2); |
| } |
| if m.get_term() != wterm { |
| panic!("#{}.{}: term = {}, want {}", j, i, m.get_term(), wterm); |
| } |
| let windex = ents.last().unwrap().get_index(); |
| let wlogterm = ents.last().unwrap().get_term(); |
| if m.get_index() != windex { |
| panic!("#{}.{}: index = {}, want {}", j, i, m.get_index(), windex); |
| } |
| if m.get_log_term() != wlogterm { |
| panic!( |
| "#{}.{}: log_term = {}, want {}", |
| j, |
| i, |
| m.get_log_term(), |
| wlogterm |
| ); |
| } |
| } |
| } |
| } |
| |
| // test_voter tests the voter denies its vote if its own log is more up-to-date |
| // than that of the candidate. |
| // Reference: section 5.4.1 |
| #[test] |
| fn test_voter() { |
| setup_for_test(); |
| let mut tests = vec![ |
| // same logterm |
| (vec![empty_entry(1, 1)], 1, 1, false), |
| (vec![empty_entry(1, 1)], 1, 2, false), |
| (vec![empty_entry(1, 1), empty_entry(1, 2)], 1, 1, true), |
| // candidate higher logterm |
| (vec![empty_entry(1, 1)], 2, 1, false), |
| (vec![empty_entry(1, 1)], 2, 2, false), |
| (vec![empty_entry(1, 1), empty_entry(1, 2)], 2, 1, false), |
| // voter higher logterm |
| (vec![empty_entry(2, 1)], 1, 1, true), |
| (vec![empty_entry(2, 1)], 1, 2, true), |
| (vec![empty_entry(2, 1), empty_entry(1, 2)], 1, 1, true), |
| ]; |
| for (i, (ents, log_term, index, wreject)) in tests.drain(..).enumerate() { |
| let s = new_storage(); |
| s.wl().append(&ents).expect(""); |
| let mut r = new_test_raft(1, vec![1, 2], 10, 1, s); |
| |
| let mut m = new_message(2, 1, MessageType::MsgRequestVote, 0); |
| m.set_term(3); |
| m.set_log_term(log_term); |
| m.set_index(index); |
| r.step(m).expect(""); |
| |
| let msgs = r.read_messages(); |
| if msgs.len() != 1 { |
| panic!("#{}: msg count = {}, want {}", i, msgs.len(), 1); |
| } |
| if msgs[0].get_msg_type() != MessageType::MsgRequestVoteResponse { |
| panic!( |
| "#{}: msg_type = {:?}, want {:?}", |
| i, |
| msgs[0].get_msg_type(), |
| MessageType::MsgRequestVoteResponse |
| ); |
| } |
| if msgs[0].get_reject() != wreject { |
| panic!( |
| "#{}: reject = {}, want {}", |
| i, |
| msgs[0].get_reject(), |
| wreject |
| ); |
| } |
| } |
| } |
| |
| // TestLeaderOnlyCommitsLogFromCurrentTerm tests that only log entries from the leader’s |
| // current term are committed by counting replicas. |
| // Reference: section 5.4.2 |
| #[test] |
| fn test_leader_only_commits_log_from_current_term() { |
| setup_for_test(); |
| let ents = vec![empty_entry(1, 1), empty_entry(2, 2)]; |
| let mut tests = vec![ |
| // do not commit log entries in previous terms |
| (1, 0), |
| (2, 0), |
| // commit log in current term |
| (3, 3), |
| ]; |
| for (i, (index, wcommit)) in tests.drain(..).enumerate() { |
| let store = new_storage(); |
| store.wl().append(&ents).expect(""); |
| let mut r = new_test_raft(1, vec![1, 2], 10, 1, store); |
| r.load_state(&hard_state(2, 0, 0)); |
| // become leader at term 3 |
| r.become_candidate(); |
| r.become_leader(); |
| r.read_messages(); |
| // propose a entry to current term |
| r.step(new_message(1, 1, MessageType::MsgPropose, 1)) |
| .expect(""); |
| |
| let mut m = new_message(2, 1, MessageType::MsgAppendResponse, 0); |
| m.set_term(r.term); |
| m.set_index(index); |
| r.step(m).expect(""); |
| if r.raft_log.committed != wcommit { |
| panic!( |
| "#{}: commit = {}, want {}", |
| i, r.raft_log.committed, wcommit |
| ); |
| } |
| } |
| } |
