refactor: refactor shutdown test

add retry count for client fix curp group recovery Signed-off-by: themanforfree <[email protected]>
xline-kv · Sep 8, 2023 · 6e4769f · 6e4769f
1 parent c41a25a
commit 6e4769f
Show file tree

Hide file tree

Showing 39 changed files with 558 additions and 308 deletions.
diff --git a/USAGE.md b/USAGE.md
@@ -38,7 +38,7 @@ auth_public_key = '/etc/xline/public_key.pem'
 auth_private_key = '/etc/xline/private_key.pem'
 ```
 
-For tuning and development purpose, the cluster section provides two subsections, curp_cfg, and client_timeout, with the following definitions and default values.
+For tuning and development purpose, the cluster section provides two subsections, curp_cfg, and client_config, with the following definitions and default values.
 
 ```toml
 [cluster.curp_config]
@@ -53,7 +53,7 @@ candidate_timeout_ticks = 2     # if a candidate cannot win an election, it will
                                 # after `candidate_timeout_ticks` ticks. Its default value is 2
 
 
-[cluster.client_timeout]
+[cluster.client_config]
 propose_timeout = '1s'          # client propose timeout
 wait_synced_timeout = '2s'      # client wait synced timeout
 retry_timeout = '50ms'          # the rpc retry interval, of which the default is 50ms

diff --git a/benchmark/src/runner.rs b/benchmark/src/runner.rs
@@ -19,7 +19,7 @@ use tokio::{
     time::{Duration, Instant},
 };
 use tracing::debug;
-use utils::config::ClientTimeout;
+use utils::config::ClientConfig;
 use xline::client::{kv_types::PutRequest, Client};
 
 use crate::{args::Commands, Benchmark};
@@ -162,10 +162,11 @@ impl CommandRunner {
             let client = Client::new(
                 self.args.endpoints.clone(),
                 self.args.use_curp,
-                ClientTimeout::new(
+                ClientConfig::new(
                     Duration::from_secs(10),
                     Duration::from_secs(5),
                     Duration::from_millis(250),
+                    3,
                 ),
             )
             .await?;

diff --git a/curp-external-api/src/cmd.rs b/curp-external-api/src/cmd.rs
@@ -155,6 +155,9 @@ where
         prepare_res: C::PR,
     ) -> Result<C::ASR, C::Error>;
 
+    /// Set the index of the last log entry that has been successfully applied to the command executor
+    fn set_last_applied(&self, index: LogIndex) -> Result<(), C::Error>;
+
     /// Index of the last log entry that has been successfully applied to the command executor
     fn last_applied(&self) -> Result<LogIndex, C::Error>;
 

diff --git a/curp-external-api/src/lib.rs b/curp-external-api/src/lib.rs
@@ -139,5 +139,3 @@ pub type LogIndex = u64;
 pub mod cmd;
 /// The command to be executed
 pub mod role_change;
-/// Snapshot
-pub mod snapshot;
diff --git a/curp-external-api/src/snapshot.rs b/curp-external-api/src/snapshot.rs
diff --git a/curp-test-utils/src/lib.rs b/curp-test-utils/src/lib.rs
@@ -14,6 +14,7 @@ pub mod test_cmd;
 
 pub const TEST_TABLE: &str = "test";
 pub const REVISION_TABLE: &str = "revision";
+pub const META_TABLE: &str = "meta";
 
 #[derive(Default, Debug)]
 pub struct TestRoleChange {

diff --git a/curp-test-utils/src/test_cmd.rs b/curp-test-utils/src/test_cmd.rs
@@ -8,7 +8,6 @@ use std::{
 };
 
 use async_trait::async_trait;
-use clippy_utilities::NumericCast;
 use curp_external_api::{
     cmd::{Command, CommandExecutor, ConflictCheck, PbCodec, ProposeId},
     LogIndex,
@@ -20,8 +19,12 @@ use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use tokio::{sync::mpsc, time::sleep};
 use tracing::debug;
+use utils::config::StorageConfig;
 
-use crate::{REVISION_TABLE, TEST_TABLE};
+use crate::{META_TABLE, REVISION_TABLE, TEST_TABLE};
+
+pub(crate) const APPLIED_INDEX_KEY: &str = "applied_index";
+pub(crate) const LAST_REVISION_KEY: &str = "last_revision";
 
 static NEXT_ID: Lazy<AtomicU64> = Lazy::new(|| AtomicU64::new(1));
 
@@ -234,7 +237,6 @@ impl PbCodec for TestCommand {
 #[derive(Debug, Clone)]
 pub struct TestCE {
     server_id: String,
-    last_applied: Arc<AtomicU64>,
     revision: Arc<AtomicI64>,
     pub store: Arc<Engine>,
     exe_sender: mpsc::UnboundedSender<(TestCommand, TestCommandResult)>,
@@ -249,7 +251,16 @@ impl CommandExecutor<TestCommand> for TestCE {
         _index: LogIndex,
     ) -> Result<<TestCommand as Command>::PR, <TestCommand as Command>::Error> {
         let rev = if let TestCommandType::Put(_) = cmd.cmd_type {
-            self.revision.fetch_add(1, Ordering::Relaxed)
+            let rev = self.revision.fetch_add(1, Ordering::Relaxed);
+            let wr_ops = vec![WriteOperation::new_put(
+                META_TABLE,
+                LAST_REVISION_KEY.into(),
+                rev.to_le_bytes().to_vec(),
+            )];
+            self.store
+                .write_batch(wr_ops, true)
+                .map_err(|e| ExecuteError(e.to_string()))?;
+            rev
         } else {
             -1
         };
@@ -315,6 +326,11 @@ impl CommandExecutor<TestCommand> for TestCE {
         self.after_sync_sender
             .send((cmd.clone(), index))
             .expect("failed to send after sync msg");
+        let mut wr_ops = vec![WriteOperation::new_put(
+            META_TABLE,
+            APPLIED_INDEX_KEY.into(),
+            index.to_le_bytes().to_vec(),
+        )];
         if let TestCommandType::Put(v) = cmd.cmd_type {
             debug!(
                 "cmd {:?}-{} revision is {}",
@@ -328,19 +344,22 @@ impl CommandExecutor<TestCommand> for TestCE {
                 .iter()
                 .map(|k| k.to_be_bytes().to_vec())
                 .collect_vec();
-            let wr_ops = keys
-                .clone()
-                .into_iter()
-                .map(|key| WriteOperation::new_put(TEST_TABLE, key, value.clone()))
-                .chain(keys.into_iter().map(|key| {
-                    WriteOperation::new_put(REVISION_TABLE, key, revision.to_be_bytes().to_vec())
-                }))
-                .collect();
+            wr_ops.extend(
+                keys.clone()
+                    .into_iter()
+                    .map(|key| WriteOperation::new_put(TEST_TABLE, key, value.clone()))
+                    .chain(keys.into_iter().map(|key| {
+                        WriteOperation::new_put(
+                            REVISION_TABLE,
+                            key,
+                            revision.to_be_bytes().to_vec(),
+                        )
+                    })),
+            );
             self.store
                 .write_batch(wr_ops, true)
                 .map_err(|e| ExecuteError(e.to_string()))?;
         }
-        self.last_applied.store(index, Ordering::Relaxed);
         debug!(
             "{} after sync cmd({:?} - {}), index: {index}",
             self.server_id,
@@ -350,8 +369,27 @@ impl CommandExecutor<TestCommand> for TestCE {
         Ok(index.into())
     }
 
+    fn set_last_applied(&self, index: LogIndex) -> Result<(), <TestCommand as Command>::Error> {
+        let ops = vec![WriteOperation::new_put(
+            META_TABLE,
+            APPLIED_INDEX_KEY.into(),
+            index.to_le_bytes().to_vec(),
+        )];
+        self.store
+            .write_batch(ops, true)
+            .map_err(|e| ExecuteError(e.to_string()))?;
+        Ok(())
+    }
+
     fn last_applied(&self) -> Result<LogIndex, ExecuteError> {
-        Ok(self.last_applied.load(Ordering::Relaxed))
+        let Some(index) = self
+            .store
+            .get(META_TABLE, APPLIED_INDEX_KEY)
+            .map_err(|e| ExecuteError(e.to_string()))? else {
+            return Ok(0);
+        };
+        let index = LogIndex::from_le_bytes(index.as_slice().try_into().unwrap());
+        Ok(index)
     }
 
     async fn snapshot(&self) -> Result<Snapshot, <TestCommand as Command>::Error> {
@@ -365,15 +403,20 @@ impl CommandExecutor<TestCommand> for TestCE {
         snapshot: Option<(Snapshot, LogIndex)>,
     ) -> Result<(), <TestCommand as Command>::Error> {
         let Some((mut snapshot, index)) = snapshot else {
-            self.last_applied.store(0, Ordering::Relaxed);
-            let ops = vec![WriteOperation::new_delete_range(TEST_TABLE, &[], &[0xff])];
+            let ops = vec![WriteOperation::new_delete_range(TEST_TABLE, &[], &[0xff]),WriteOperation::new_delete(META_TABLE, APPLIED_INDEX_KEY.as_ref())];
             self.store
                 .write_batch(ops, true)
                 .map_err(|e| ExecuteError(e.to_string()))?;
             return Ok(());
         };
-        self.last_applied
-            .store(index.numeric_cast(), Ordering::Relaxed);
+        let ops = vec![WriteOperation::new_put(
+            META_TABLE,
+            APPLIED_INDEX_KEY.into(),
+            index.to_le_bytes().to_vec(),
+        )];
+        self.store
+            .write_batch(ops, true)
+            .map_err(|e| ExecuteError(e.to_string()))?;
         snapshot.rewind().unwrap();
         self.store
             .apply_snapshot(snapshot, &[TEST_TABLE])
@@ -388,14 +431,24 @@ impl TestCE {
         server_id: String,
         exe_sender: mpsc::UnboundedSender<(TestCommand, TestCommandResult)>,
         after_sync_sender: mpsc::UnboundedSender<(TestCommand, LogIndex)>,
+        storage_cfg: StorageConfig,
     ) -> Self {
+        let engine_type = match storage_cfg {
+            StorageConfig::Memory => EngineType::Memory,
+            StorageConfig::RocksDB(path) => EngineType::Rocks(path),
+            _ => unreachable!("Not supported storage type"),
+        };
+        let store =
+            Arc::new(Engine::new(engine_type, &[TEST_TABLE, REVISION_TABLE, META_TABLE]).unwrap());
+        let rev = store
+            .get(META_TABLE, LAST_REVISION_KEY)
+            .unwrap()
+            .map(|r| i64::from_le_bytes(r.as_slice().try_into().unwrap()))
+            .unwrap_or(0);
         Self {
             server_id,
-            last_applied: Arc::new(AtomicU64::new(0)),
-            revision: Arc::new(AtomicI64::new(1)),
-            store: Arc::new(
-                Engine::new(EngineType::Memory, &[TEST_TABLE, REVISION_TABLE]).unwrap(),
-            ),
+            revision: Arc::new(AtomicI64::new(rev + 1)),
+            store,
             exe_sender,
             after_sync_sender,
         }

diff --git a/curp/proto/error.proto b/curp/proto/error.proto
@@ -13,6 +13,7 @@ message ProposeError {
         string encode_error = 4;
         google.protobuf.Empty not_leader = 5;
         google.protobuf.Empty shutdown = 6;
+        google.protobuf.Empty timeout = 7;
     }
 }
 

diff --git a/curp/proto/message.proto b/curp/proto/message.proto
@@ -94,7 +94,7 @@ service Protocol {
     rpc WaitSynced(commandpb.WaitSyncedRequest)
         returns (commandpb.WaitSyncedResponse);
     rpc Shutdown(ShutdownRequest) returns (ShutdownResponse);
-        rpc AppendEntries(AppendEntriesRequest) returns (AppendEntriesResponse);
+    rpc AppendEntries(AppendEntriesRequest) returns (AppendEntriesResponse);
     rpc Vote(VoteRequest) returns (VoteResponse);
     rpc InstallSnapshot(stream InstallSnapshotRequest)
         returns (InstallSnapshotResponse);