tests: use fail point (#114)

* tests: use fail point

Remove the hacky function pointer, and use failpoint to do the precise
control.

* Add note why failpoints is seperate

* cargo fmt

.travis.yml

@@ -27,3 +27,5 @@ script:
   - cargo test --all -- --nocapture
   # Validate benches still work.
   - cargo bench --all -- --test
+  # Because failpoints inject failure in code path, which will affect all concurrently running tests, Hence they need to be synchronized, which make tests slow.
+  - cargo test --tests --features failpoint -- --nocapture

Cargo.toml

@@ -11,16 +11,22 @@ documentation = "https://docs.rs/raft"
 description = "The rust language implementation of Raft algorithm."
 categories = ["algorithms", "database-implementations"]
 
+[features]
+default = []
+failpoint = ["fail"]
+
 [dependencies]
 log = ">0.2"
 protobuf = "2.0.4"
 quick-error = "1.2.2"
 rand = "0.5.4"
 fxhash = "0.2.1"
+fail = { version = "0.2", optional = true }
 
 [dev-dependencies]
 env_logger = "0.5"
 criterion = ">0.2.4"
+lazy_static = "1.0"
 
 [[bench]]
 name = "benches"

examples/single_mem_node/main.rs

@@ -196,8 +196,7 @@ fn send_propose(sender: mpsc::Sender<Msg>) {
                 cb: Box::new(move || {
                     s1.send(0).unwrap();
                 }),
-            })
-            .unwrap();
+            }).unwrap();
 
         let n = r1.recv().unwrap();
         assert_eq!(n, 0);

src/lib.rs

@@ -264,6 +264,9 @@ For more information, check out an [example](examples/single_mem_node/main.rs#L1
 #![cfg_attr(feature = "cargo-clippy", feature(tool_lints))]
 #![deny(missing_docs)]
 
+#[cfg(feature = "failpoint")]
+#[macro_use]
+extern crate fail;
 extern crate fxhash;
 #[macro_use]
 extern crate log;

src/raft.rs

@@ -176,11 +176,6 @@ pub struct Raft<T: Storage> {
     min_election_timeout: usize,
     max_election_timeout: usize,
 
-    /// Will be called when step** is about to be called.
-    /// return false will skip step**. Only used for test purpose.
-    #[doc(hidden)]
-    pub before_step_state: Option<Box<FnMut(&Message) -> bool + Send>>,
-
     /// Tag is only used for logging
     tag: String,
 }
@@ -264,7 +259,6 @@ impl<T: Storage> Raft<T> {
             term: Default::default(),
             election_elapsed: Default::default(),
             pending_conf_index: Default::default(),
-            before_step_state: None,
             vote: Default::default(),
             heartbeat_elapsed: Default::default(),
             randomized_election_timeout: 0,
@@ -578,7 +572,10 @@ impl<T: Storage> Raft<T> {
         self.bcast_heartbeat_with_ctx(ctx)
     }
 
-    #[cfg_attr(feature = "cargo-clippy", allow(clippy::needless_pass_by_value))]
+    #[cfg_attr(
+        feature = "cargo-clippy",
+        allow(clippy::needless_pass_by_value)
+    )]
     fn bcast_heartbeat_with_ctx(&mut self, ctx: Option<Vec<u8>>) {
         let self_id = self.id;
         let mut prs = self.take_prs();
@@ -1001,12 +998,8 @@ impl<T: Storage> Raft<T> {
             return Ok(());
         }
 
-        if let Some(ref mut f) = self.before_step_state {
-            if !f(&m) {
-                // skip step**
-                return Ok(());
-            }
-        }
+        #[cfg(feature = "failpoint")]
+        fail_point!("before_step");
 
         match m.get_msg_type() {
             MessageType::MsgHup => if self.state != StateRole::Leader {
@@ -1016,8 +1009,7 @@ impl<T: Storage> Raft<T> {
                         self.raft_log.applied + 1,
                         self.raft_log.committed + 1,
                         raft_log::NO_LIMIT,
-                    )
-                    .expect("unexpected error getting unapplied entries");
+                    ).expect("unexpected error getting unapplied entries");
                 let n = self.num_pending_conf(&ents);
                 if n != 0 && self.raft_log.committed > self.raft_log.applied {
                     warn!(

src/raw_node.rs

@@ -286,7 +286,10 @@ impl<T: Storage> RawNode<T> {
     }
 
     /// ProposeConfChange proposes a config change.
-    #[cfg_attr(feature = "cargo-clippy", allow(clippy::needless_pass_by_value))]
+    #[cfg_attr(
+        feature = "cargo-clippy",
+        allow(clippy::needless_pass_by_value)
+    )]
     pub fn propose_conf_change(&mut self, context: Vec<u8>, cc: ConfChange) -> Result<()> {
         let data = protobuf::Message::write_to_bytes(&cc)?;
         let mut m = Message::new();

src/util.rs

@@ -64,8 +64,7 @@ pub fn limit_size<T: Message + Clone>(entries: &mut Vec<T>, max: u64) {
                 size += u64::from(Message::compute_size(e));
                 size <= max
             }
-        })
-        .count();
+        }).count();
 
     entries.truncate(limit);
 }

tests/failpoint_cases/mod.rs

@@ -0,0 +1,63 @@
+// Copyright 2018 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.
+
+use fail;
+use raft::eraftpb::MessageType;
+use std::sync::*;
+use test_util::*;
+
+lazy_static! {
+    /// Failpoints are global structs, hence rules set in different cases
+    /// may affect each other. So use a global lock to synchronize them.
+    static ref LOCK: Mutex<()> = {
+        Mutex::new(())
+    };
+}
+
+fn setup<'a>() -> MutexGuard<'a, ()> {
+    setup_for_test();
+    // We don't want a failed test breaks others.
+    let guard = LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    fail::teardown();
+    fail::setup();
+    guard
+}
+
+// test_reject_stale_term_message tests that if a server receives a request with
+// a stale term number, it rejects the request.
+// Our implementation ignores the request instead.
+// Reference: section 5.1
+#[test]
+fn test_reject_stale_term_message() {
+    let _guard = setup();
+    let mut r = new_test_raft(1, vec![1, 2, 3], 10, 1, new_storage());
+    fail::cfg("before_step", "panic").unwrap();
+    r.load_state(&hard_state(2, 0, 0));
+
+    let mut m = new_message(0, 0, MessageType::MsgAppend, 0);
+    m.set_term(r.term - 1);
+    r.step(m).expect("");
+}
+
+// ensure that the Step function ignores the message from old term and does not pass it to the
+// actual stepX function.
+#[test]
+fn test_step_ignore_old_term_msg() {
+    let _guard = setup();
+    let mut sm = new_test_raft(1, vec![1], 10, 1, new_storage());
+    fail::cfg("before_step", "panic").unwrap();
+    sm.term = 2;
+    let mut m = new_message(0, 0, MessageType::MsgAppend, 0);
+    m.set_term(1);
+    sm.step(m).expect("");
+}