Merge branch 'master' into persistence-2-consensus

Author: dinowernli

src/harness.rs

@@ -30,6 +30,7 @@ use crate::raft::{
 // production deployment, these participants might be on different actual machines,
 // but the harness manages all of them in a single process for convenience.
 pub struct Harness {
+    cluster_name: String,
     instances: Vec<Instance>,
     diagnostics: Arc<Mutex<Diagnostics>>,
     failures: Arc<Mutex<FailureOptions>>,
@@ -40,6 +41,7 @@ pub struct Harness {
 // addresses, and then start the underlying services (which require knowledge of the
 // ports for the other participants).
 pub struct HarnessBuilder {
+    cluster_name: String,
     bound: Vec<BoundAddress>,
     failure: FailureOptions,
     options: Options,
@@ -50,7 +52,6 @@ impl HarnessBuilder {
     // of the serving process for all instances managed by the harness.
     pub async fn build(
         self,
-        cluster_name: &str,
         wipe_persistence: bool,
     ) -> Result<(Harness, Pin<Box<dyn Future<Output = ()> + Send>>), Box<dyn Error>> {
         let diag = Arc::new(Mutex::new(Diagnostics::new()));
@@ -68,7 +69,7 @@ impl HarnessBuilder {
             let persistence_path = env::temp_dir()
                 .as_path()
                 .join("concord")
-                .join(&cluster_name)
+                .join(&self.cluster_name)
                 .join(&address.name);
 
             let options = raft_options.clone().with_persistence(
@@ -95,6 +96,7 @@ impl HarnessBuilder {
             join_all(serving).await;
         });
         let harness = Harness {
+            cluster_name: self.cluster_name,
             instances,
             diagnostics: diag,
             failures,
@@ -105,6 +107,7 @@ impl HarnessBuilder {
     // Consumes this instance and returns an instance with the failure options set.
     pub fn with_failure(self: Self, failure_options: FailureOptions) -> Self {
         Self {
+            cluster_name: self.cluster_name,
             bound: self.bound,
             failure: failure_options,
             options: self.options,
@@ -114,6 +117,7 @@ impl HarnessBuilder {
     // Consumes this instance and returns an instance with the raft options set.
     pub fn with_options(self: Self, options: Options) -> Self {
         Self {
+            cluster_name: self.cluster_name,
             bound: self.bound,
             failure: self.failure,
             options,
@@ -129,15 +133,19 @@ impl HarnessBuilder {
 impl Harness {
     // Creates a harness builder. This will immediately bind an incoming port for
     // each supplied instance name.
-    pub async fn builder(names: Vec<String>) -> Result<HarnessBuilder, Box<dyn Error>> {
+    pub async fn builder(
+        cluster_name: &str,
+        server_names: &[&str],
+    ) -> Result<HarnessBuilder, Box<dyn Error>> {
         let mut bound = Vec::new();
-        for name in names {
+        for &name in server_names {
             let listener = TcpListener::bind("[::1]:0").await?;
             let port = listener.local_addr()?.port();
-            let server = server("::1", port as i32, name.as_str());
+            let server = server("::1", port as i32, name);
             bound.push(BoundAddress { listener, server })
         }
         Ok(HarnessBuilder {
+            cluster_name: cluster_name.to_string(),
             bound,
             failure: FailureOptions::no_failures(),
 
@@ -147,6 +155,10 @@ impl Harness {
         })
     }
 
+    pub fn name(&self) -> &str {
+        &self.cluster_name
+    }
+
     // Returns all the addresses managed by this harness. Note that this can include
     // Raft members that are not currently part of the cluster.
     pub fn addresses(&self) -> Vec<Server> {
@@ -178,13 +190,13 @@ impl Harness {
 
     // Update the cluster membership to contain only the supplied members. The supplied
     // members must all be valid servers as defined at harness creation time.
-    pub async fn update_members(&self, members: Vec<String>) -> Result<(), Status> {
+    pub async fn update_members(&self, members: Vec<&str>) -> Result<(), Status> {
         let client = self.make_raft_client();
 
         let addresses = self.addresses().clone();
         let mut new_members: Vec<Server> = Vec::new();
         for m in members {
-            let server = addresses.iter().find(|&a| &a.name == &m);
+            let server = addresses.iter().find(|&a| &a.name == m);
             match server {
                 None => return Err(Status::invalid_argument(format!("unknown server: {}", &m))),
                 Some(s) => new_members.push(s.clone()),
@@ -363,7 +375,7 @@ impl Instance {
                 failure_options,
             )
             .await
-            .map_err(|e| format!("Failed to create Raft for '{}': {}", address.name, e))?,
+            .map_err(|e| format!("Failed to create raft impl for '{}': {}", address.name, e))?,
         );
         let raft_grpc = RaftServer::from_arc(raft.clone());
 

src/integration_test.rs

@@ -6,15 +6,11 @@ use std::time::Duration;
 
 const TIMEOUT: Duration = Duration::from_secs(3);
 const CLUSTER_NAME: &str = "test-cluster";
-
-fn names() -> Vec<String> {
-    vec!["A".to_string(), "B".to_string(), "C".to_string()]
-}
+const NAMES: [&str; 3] = ["A", "B", "C"];
 
 #[tokio::test]
 async fn test_start_and_elect_leader() {
-    let harness = make_harness(names()).await;
-
+    let harness = make_harness(&NAMES).await;
     harness.wait_for_leader(TIMEOUT, term_greater(0)).await;
 
     harness.validate().await;
@@ -25,7 +21,8 @@ async fn test_start_and_elect_leader() {
 async fn test_start_and_elect_leader_many_nodes() {
     let n = 17;
     let owned: Vec<String> = (1..=n).map(|i| i.to_string()).collect();
-    let harness = make_harness(owned).await;
+    let names: Vec<&str> = owned.iter().map(|s| s.as_str()).collect();
+    let harness = make_harness(&names).await;
 
     harness.wait_for_leader(TIMEOUT, term_greater(0)).await;
 
@@ -35,7 +32,7 @@ async fn test_start_and_elect_leader_many_nodes() {
 
 #[tokio::test]
 async fn test_disconnect_leader() {
-    let harness = make_harness(names()).await;
+    let harness = make_harness(&NAMES).await;
 
     // Wait for the initial leader and capture its term and server.
     let (term1, leader1) = harness.wait_for_leader(TIMEOUT, term_greater(0)).await;
@@ -65,7 +62,7 @@ async fn test_disconnect_leader() {
 
 #[tokio::test]
 async fn test_commit() {
-    let harness = make_harness(names()).await;
+    let harness = make_harness(&NAMES).await;
     harness.wait_for_leader(TIMEOUT, term_greater(0)).await;
     let client = harness.make_raft_client();
 
@@ -79,42 +76,35 @@ async fn test_commit() {
 
 #[tokio::test]
 async fn test_reconfigure_cluster() {
-    let names = vec![
-        "A".to_string(),
-        "B".to_string(),
-        "C".to_string(),
-        "D".to_string(),
-        "E".to_string(),
-    ];
-    let harness = make_harness(names.clone()).await;
+    let names = vec!["A", "B", "C", "D", "E"];
+    let harness = make_harness(&names).await;
 
     let (t1, leader1) = harness.wait_for_leader(TIMEOUT, term_greater(0)).await;
 
-    let without_leader: Vec<String> = names
-        .clone()
+    let without_leader: Vec<&str> = names
         .iter()
-        .filter(|&s| *s != leader1.name)
-        .map(|s| s.clone())
+        .copied()
+        .filter(|s| *s != leader1.name)
         .collect();
     assert_eq!(without_leader.len(), 4);
 
     // Change cluster to contain only 3 members, and not including the current leader.
-    let new_members: Vec<String> = without_leader.iter().take(3).map(|s| s.clone()).collect();
+    let new_members: Vec<&str> = without_leader.iter().copied().take(3).collect();
     let result = harness.update_members(new_members.clone()).await;
     assert!(result.is_ok());
 
     // Wait for a new leader and verify.
     let (_, leader2) = harness.wait_for_leader(TIMEOUT, term_greater(t1)).await;
     assert_ne!(&leader2.name, &leader1.name);
-    assert!(new_members.contains(&leader2.name));
+    assert!(new_members.contains(&leader2.name.as_str()));
 
     harness.validate().await;
     harness.stop().await;
 }
 
 #[tokio::test]
 async fn test_keyvalue() {
-    let harness = make_harness(names()).await;
+    let harness = make_harness(&NAMES).await;
     let mut kv = harness.make_kv_client().await;
 
     let k1 = "k1".as_bytes().to_vec();
@@ -137,7 +127,7 @@ async fn test_keyvalue() {
 async fn test_snapshotting() {
     let raft_options =
         Options::new_without_persistence_for_testing().with_compaction(5 * 1024 * 1024, 1000);
-    let harness = make_harness_with_options(names(), Some(raft_options)).await;
+    let harness = make_harness_with_options(&NAMES, Some(raft_options)).await;
 
     // Disconnect a node that will later have to catch up.
     harness.failures().lock().await.disconnect("B");
@@ -175,22 +165,21 @@ fn term_greater(n: i64) -> Box<dyn Fn(&(i64, Server)) -> bool> {
     Box::new(move |(term, _)| *term > n)
 }
 
-async fn make_harness(nodes: Vec<String>) -> Harness {
+async fn make_harness(nodes: &[&str]) -> Harness {
     make_harness_with_options(nodes, None).await
 }
 
-async fn make_harness_with_options(nodes: Vec<String>, options: Option<Options>) -> Harness {
-    let mut builder = Harness::builder(nodes).await.expect("builder");
+async fn make_harness_with_options(nodes: &[&str], options: Option<Options>) -> Harness {
+    let mut builder = Harness::builder(CLUSTER_NAME, nodes)
+        .await
+        .expect("builder");
 
     if let Some(opts) = options {
         builder = builder.with_options(opts)
     }
 
     let wipe_persistence = true;
-    let (harness, serving) = builder
-        .build(CLUSTER_NAME, wipe_persistence)
-        .await
-        .expect("harness");
+    let (harness, serving) = builder.build(wipe_persistence).await.expect("harness");
     harness.start().await;
     tokio::spawn(async { serving.await });
     harness

src/main.rs

@@ -226,16 +226,6 @@ async fn run_reconfigure_loop(
     info!("Finished")
 }
 
-fn names() -> Vec<String> {
-    vec![
-        "A".to_string(),
-        "B".to_string(),
-        "C".to_string(),
-        "D".to_string(),
-        "E".to_string(),
-    ]
-}
-
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     // This allows configuring the filters using the RUST_LOG env variable.
@@ -249,14 +239,16 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .init();
     let arguments = Arc::new(Arguments::from_args());
 
-    let (harness, serving) = Harness::builder(names())
+    let (harness, serving) = Harness::builder(CLUSTER_NAME, &["A", "B", "C", "D", "E"])
         .await
         .expect("builder")
         .with_failure(make_default_failure_options())
-        .build(CLUSTER_NAME, arguments.wipe_persistence)
+        .build(arguments.wipe_persistence)
         .await
         .expect("harness");
 
+    info!("Created cluster '{}'", harness.name());
+
     let addresses = harness.addresses();
     let diagnostics = harness.diagnostics();
 

src/raft/consensus.rs

@@ -132,8 +132,6 @@ impl RaftImpl {
         options: Options,
         failures: Arc<Mutex<FailureOptions>>,
     ) -> RaftResult<RaftImpl> {
-        let snapshot_bytes = state_machine.lock().await.create_snapshot();
-
         // A config which gives us (once applied) all the initial members as voters.
         let initial_cluster_config = ClusterConfig {
             voters: all.clone().to_vec(),

src/raft/store.rs

@@ -2,9 +2,7 @@ use crate::raft::diagnostics::ServerDiagnostics;
 use crate::raft::error::RaftError::Initialization;
 use crate::raft::error::RaftResult;
 use crate::raft::log::LogSlice;
-use crate::raft::persistence::{
-    Persistence, PersistenceError, PersistenceOptions, PersistentState,
-};
+use crate::raft::persistence::{Persistence, PersistenceOptions, PersistentState};
 use crate::raft::raft_common_proto::entry::Data;
 use crate::raft::raft_common_proto::entry::Data::Config;
 use crate::raft::raft_common_proto::{ClusterConfig, Entry, EntryId, Server};
@@ -14,10 +12,7 @@ use bytes::Bytes;
 use futures::channel::oneshot::{Receiver, Sender, channel};
 use std::cmp::Ordering;
 use std::collections::BTreeSet;
-use std::error::Error;
-use std::fmt::{Display, Formatter, Write};
 use std::hash::{Hash, Hasher};
-use thiserror::Error;
 use tonic::{Code, Status};
 use tracing::{debug, info, warn};