Merge branch 'master' into persistence-2-consensus

Author: dinowernli

src/harness.rs

@@ -7,6 +7,8 @@ use futures::Future;
 use futures::future::join_all;
 use std::env;
 use std::error::Error;
+use std::fs::remove_dir_all;
+use std::path::PathBuf;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
@@ -34,6 +36,7 @@ pub struct Harness {
     instances: Vec<Instance>,
     diagnostics: Arc<Mutex<Diagnostics>>,
     failures: Arc<Mutex<FailureOptions>>,
+    persistence_dir: String,
 }
 
 // Used to capture the intermediate state while building a harness. This is necessary
@@ -62,20 +65,22 @@ impl HarnessBuilder {
         let mut instances = Vec::new();
 
         let raft_options = self.options;
+
+        // Persist at something like /tmp/concord/<cluster>
+        let cluster_dir: PathBuf = env::temp_dir()
+            .as_path()
+            .join("concord")
+            .join(&self.cluster_name);
+
         for bound in self.bound {
             let (address, listener) = (bound.server, bound.listener);
 
-            // Use something like /tmp/concord/<cluster>/<server> for persistence
-            let persistence_path = env::temp_dir()
-                .as_path()
-                .join("concord")
-                .join(&self.cluster_name)
-                .join(&address.name);
+            // Use <cluster-dir>/<server-name> for persistence
+            let instance_dir = cluster_dir.clone().join(&address.name);
 
-            let options = raft_options.clone().with_persistence(
-                persistence_path.to_str().unwrap().as_ref(),
-                wipe_persistence,
-            );
+            let options = raft_options
+                .clone()
+                .with_persistence(instance_dir.to_str().unwrap().as_ref(), wipe_persistence);
 
             let (instance, future) = Instance::new(
                 &address,
@@ -100,6 +105,7 @@ impl HarnessBuilder {
             instances,
             diagnostics: diag,
             failures,
+            persistence_dir: cluster_dir.into_os_string().into_string().unwrap(),
         };
         Ok((harness, future))
     }
@@ -230,6 +236,13 @@ impl Harness {
         }
     }
 
+    // Removes the directory in which the cluster's state lives (and all children).
+    pub async fn wipe_persistence(&self) {
+        let dir = self.persistence_dir.clone();
+        assert!(!dir.is_empty());
+        remove_dir_all(dir).expect("remove_dir_all");
+    }
+
     // Repeatedly makes KV requests until the supplied key has a value. Returns the result.
     #[cfg(test)]
     pub async fn wait_for_key(&self, key: &[u8], timeout_duration: Duration) -> GetResponse {

src/integration_test.rs

@@ -2,6 +2,7 @@ use crate::harness::Harness;
 use crate::keyvalue::keyvalue_proto::PutRequest;
 use crate::raft::Options;
 use crate::raft::raft_common_proto::Server;
+use rand::{Rng, distributions::Alphanumeric};
 use std::time::Duration;
 
 const TIMEOUT: Duration = Duration::from_secs(3);
@@ -15,6 +16,7 @@ async fn test_start_and_elect_leader() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -28,6 +30,7 @@ async fn test_start_and_elect_leader_many_nodes() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -58,6 +61,7 @@ async fn test_disconnect_leader() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -72,6 +76,7 @@ async fn test_commit() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -100,6 +105,7 @@ async fn test_reconfigure_cluster() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -121,6 +127,10 @@ async fn test_keyvalue() {
 
     assert_eq!(&entry.key, &k1);
     assert_eq!(&entry.value, &v1);
+
+    harness.validate().await;
+    harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 #[tokio::test]
@@ -158,6 +168,7 @@ async fn test_snapshotting() {
 
     harness.validate().await;
     harness.stop().await;
+    harness.wipe_persistence().await;
 }
 
 // Convenience method that returns a matcher for terms greater than a value.
@@ -169,8 +180,18 @@ async fn make_harness(nodes: &[&str]) -> Harness {
     make_harness_with_options(nodes, None).await
 }
 
+fn make_name() -> String {
+    let suffix: String = rand::thread_rng()
+        .sample_iter(&Alphanumeric)
+        .filter(|c| c.is_ascii_lowercase())
+        .take(4)
+        .map(char::from)
+        .collect();
+    format!("{}-{}", CLUSTER_NAME, suffix)
+}
+
 async fn make_harness_with_options(nodes: &[&str], options: Option<Options>) -> Harness {
-    let mut builder = Harness::builder(CLUSTER_NAME, nodes)
+    let mut builder = Harness::builder(make_name().as_str(), nodes)
         .await
         .expect("builder");
 

src/raft/consensus.rs

@@ -1028,18 +1028,29 @@ impl Raft for RaftImpl {
         let next_term = term + 1;
         state.reset_follower_timer(self.state.clone(), next_term);
 
-        // Make sure we have the previous log index sent. Note that COMPACTED
-        // can happen whenever we have no entries (e.g.,initially or just after
-        // a snapshot install).
-        let previous = &request.previous.ok_or(RaftError::missing("previous"))?;
-        let next_index = state.store.next_index();
-        if previous.index >= next_index {
-            // Let the leader know that this entry is too far in the future, so
-            // it can try again from an earlier index.
+        // Make sure we have the previous log index sent.
+        //
+        // This can be false in two cases:
+        // 1) The leader's "previous" is too far in the future, i.e., the index is larger
+        //    than the next index we expect to append to our log.
+        // 2) The leader's "previous" is too far in the past, i.e., the entry has been
+        //    compacted into our snapshot, and is no longer present in our log.
+        //
+        // Right now, we only handle case (1) below.
+        let leader_previous = &request.previous.ok_or(RaftError::missing("previous"))?;
+        let expected_next_index = state.store.next_index();
+
+        // The Raft paper says to return false if we can't check the exact (term, index)
+        // match of the leader's "previous", but we can't do that for case (2).
+        let conflict = state.store.conflict(leader_previous);
+
+        // Either way, we inform the leader that we were unable to append, and the leader will
+        // try again with an earlier "previous", which will hopefully match.
+        if conflict || leader_previous.index >= expected_next_index {
             return Ok(Response::new(AppendResponse {
                 term,
                 success: false,
-                next: next_index,
+                next: expected_next_index,
             }));
         }
 
@@ -1408,11 +1419,11 @@ mod tests {
         let append_request = AppendRequest {
             term: 12,
             leader: Some(leader.clone()),
-            previous: Some(entry_id(-1, -1)),
+            previous: Some(entry_id(0, 0)),
             entries: vec![
-                entry(entry_id(8, 0), Vec::new()),
                 entry(entry_id(8, 1), Vec::new()),
                 entry(entry_id(8, 2), Vec::new()),
+                entry(entry_id(8, 3), Vec::new()),
             ],
             committed: 0,
         };
@@ -1432,16 +1443,16 @@ mod tests {
             let state = raft_state.lock().await;
             assert_eq!(state.current_leader(), Some(leader.clone()));
             assert_eq!(state.term(), 12);
-            assert!(!state.store.is_index_compacted(0)); // Not compacted
-            assert_eq!(state.store.next_index(), 3);
+            assert!(!state.store.is_index_compacted(1)); // Not compacted
+            assert_eq!(state.store.next_index(), 4);
         }
 
         // Run a compaction, should have no effect
         {
             let mut state = raft_state.lock().await;
             state.store.try_compact().await;
-            assert!(!state.store.is_index_compacted(0)); // Not compacted
-            assert_eq!(state.store.next_index(), 3);
+            assert!(!state.store.is_index_compacted(1)); // Not compacted
+            assert_eq!(state.store.next_index(), 4);
         }
 
         // Now send an append request with a payload large enough to trigger compaction.
@@ -1450,14 +1461,14 @@ mod tests {
         let append_request_2 = AppendRequest {
             term: 12,
             leader: Some(leader.clone()),
-            previous: Some(entry_id(8, 2)),
+            previous: Some(entry_id(8, 3)),
             entries: vec![entry(
-                entry_id(8, 3),
+                entry_id(8, 4),
                 vec![0; 2 * compaction_bytes as usize],
             )],
             // This tells the follower that the entries are committed (only committed
             // entries are eligible for compaction).
-            committed: 3,
+            committed: 4,
         };
 
         let append_response_2 = client
@@ -1471,16 +1482,16 @@ mod tests {
         assert!(append_response_2.success);
         {
             let state = raft_state.lock().await;
-            assert!(!state.store.is_index_compacted(0)); // Not compacted
-            assert_eq!(state.store.next_index(), 4); // New entry incorporated
+            assert!(!state.store.is_index_compacted(1)); // Not compacted
+            assert_eq!(state.store.next_index(), 5); // New entry incorporated
         }
 
         // Run a compaction, this one should actually compact things now.
         {
             let mut state = raft_state.lock().await;
             state.store.try_compact().await;
-            assert!(state.store.is_index_compacted(0)); // Compacted
-            assert_eq!(state.store.get_latest_snapshot().last.index, 3)
+            assert!(state.store.is_index_compacted(1)); // Compacted
+            assert_eq!(state.store.get_latest_snapshot().last.index, 4)
         }
     }
 

src/raft/error.rs

@@ -30,6 +30,9 @@ pub enum RaftError {
     #[error("Internal error: {0}")]
     Internal(String),
 
+    #[error("Non-contiguous log entries supplied. Expected index {expected}, got {actual}")]
+    NonContiguousLog { expected: i64, actual: i64 },
+
     #[error("Stale term state")]
     StaleState,
 }