Correct lease with check quorum. (#141)

* Correct lease with check quorum.

src/config.rs

@@ -100,6 +100,8 @@ pub struct Config {
     pub max_election_tick: usize,
 
     /// Choose the linearizability mode or the lease mode to read data. If you don’t care about the read consistency and want a higher read performance, you can use the lease mode.
+    ///
+    /// Setting this to `LeaseBased` requires `check_quorum = true`.
     pub read_only_option: ReadOnlyOption,
 
     /// Don't broadcast an empty raft entry to notify follower to commit an entry.
@@ -204,6 +206,12 @@ impl Config {
             ));
         }
 
+        if self.read_only_option == ReadOnlyOption::LeaseBased && !self.check_quorum {
+            return Err(Error::ConfigInvalid(
+                "read_only_option == LeaseBased requires check_quorum == true".into(),
+            ));
+        }
+
         Ok(())
     }
 }

src/raft.rs

@@ -1406,14 +1406,11 @@ impl<T: Storage> Raft<T> {
                             self.bcast_heartbeat_with_ctx(Some(ctx));
                         }
                         ReadOnlyOption::LeaseBased => {
-                            let mut read_index = INVALID_INDEX;
-                            if self.check_quorum {
-                                read_index = self.raft_log.committed
-                            }
+                            let mut read_index = self.raft_log.committed;
                             if m.get_from() == INVALID_ID || m.get_from() == self.id {
                                 // from local member
                                 let rs = ReadState {
-                                    index: self.raft_log.committed,
+                                    index: read_index,
                                     request_ctx: m.take_entries()[0].take_data(),
                                 };
                                 self.read_states.push(rs);

tests/integration_cases/test_raft.rs

@@ -2319,7 +2319,7 @@ fn test_read_only_option_lease_without_check_quorum() {
     let read_states = &nt.peers[&2].read_states;
     assert!(!read_states.is_empty());
     let rs = &read_states[0];
-    assert_eq!(rs.index, INVALID_ID);
+    assert_eq!(rs.index, 1);
     let vec_ctx = ctx.as_bytes().to_vec();
     assert_eq!(rs.request_ctx, vec_ctx);
 }