refactor: after sync command

Signed-off-by: bsbds <69835502+bsbds@users.noreply.github.com>

chore: add reminder of a revision issue

Signed-off-by: bsbds <69835502+bsbds@users.noreply.github.com>

crates/curp-external-api/src/cmd.rs

@@ -28,7 +28,7 @@ impl<T> pri::Serializable for T where T: pri::ThreadSafe + Clone + Serialize + D
 #[async_trait]
 pub trait Command: pri::Serializable + ConflictCheck + PbCodec {
     /// Error type
-    type Error: pri::Serializable + PbCodec + std::error::Error;
+    type Error: pri::Serializable + PbCodec + std::error::Error + Clone;
 
     /// K (key) is used to tell confliction
     ///
@@ -75,24 +75,6 @@ pub trait Command: pri::Serializable + ConflictCheck + PbCodec {
     {
         <E as CommandExecutor<Self>>::execute(e, self).await
     }
-
-    /// Execute the command after_sync callback
-    ///
-    /// # Errors
-    ///
-    /// Return `Self::Error` when `CommandExecutor::after_sync` goes wrong
-    #[inline]
-    async fn after_sync<E>(
-        &self,
-        e: &E,
-        index: LogIndex,
-        prepare_res: Self::PR,
-    ) -> Result<Self::ASR, Self::Error>
-    where
-        E: CommandExecutor<Self> + Send + Sync,
-    {
-        <E as CommandExecutor<Self>>::after_sync(e, self, index, prepare_res).await
-    }
 }
 
 /// Check conflict of two keys
@@ -141,17 +123,12 @@ where
     /// This function may return an error if there is a problem executing the command.
     async fn execute(&self, cmd: &C) -> Result<C::ER, C::Error>;
 
-    /// Execute the after_sync callback
-    ///
-    /// # Errors
-    ///
-    /// This function may return an error if there is a problem executing the after_sync callback.
+    /// Batch execute the after_sync callback
     async fn after_sync(
         &self,
-        cmd: &C,
-        index: LogIndex,
-        prepare_res: C::PR,
-    ) -> Result<C::ASR, C::Error>;
+        cmds: Vec<AfterSyncCmd<'_, C>>,
+        highest_index: LogIndex,
+    ) -> Result<Vec<(C::ASR, Option<C::ER>)>, C::Error>;
 
     /// Set the index of the last log entry that has been successfully applied to the command executor
     ///
@@ -215,3 +192,29 @@ impl From<DecodeError> for PbSerializeError {
         PbSerializeError::RpcDecode(err)
     }
 }
+
+/// After sync command type
+#[derive(Debug)]
+pub struct AfterSyncCmd<'a, C> {
+    /// The command
+    cmd: &'a C,
+    /// Whether the command needs to be executed in after sync stage
+    to_exectue: bool,
+}
+
+impl<'a, C> AfterSyncCmd<'a, C> {
+    /// Creates a new `AfterSyncCmd`
+    pub fn new(cmd: &'a C, to_exectue: bool) -> Self {
+        Self { cmd, to_exectue }
+    }
+
+    /// Gets the command
+    pub fn cmd(&self) -> &'a C {
+        self.cmd
+    }
+
+    /// Convert self into parts
+    pub fn into_parts(self) -> (&'a C, bool) {
+        (self.cmd, self.to_exectue)
+    }
+}

crates/curp-test-utils/src/test_cmd.rs

@@ -9,7 +9,7 @@ use std::{
 
 use async_trait::async_trait;
 use curp_external_api::{
-    cmd::{Command, CommandExecutor, ConflictCheck, PbCodec},
+    cmd::{AfterSyncCmd, Command, CommandExecutor, ConflictCheck, PbCodec},
     InflightId, LogIndex,
 };
 use engine::{
@@ -307,51 +307,72 @@ impl CommandExecutor<TestCommand> for TestCE {
 
     async fn after_sync(
         &self,
-        cmd: &TestCommand,
-        index: LogIndex,
-        revision: <TestCommand as Command>::PR,
-    ) -> Result<<TestCommand as Command>::ASR, <TestCommand as Command>::Error> {
-        sleep(cmd.as_dur).await;
-        if cmd.as_should_fail {
+        cmds: Vec<AfterSyncCmd<'_, TestCommand>>,
+        highest_index: LogIndex,
+    ) -> Result<
+        Vec<(
+            <TestCommand as Command>::ASR,
+            Option<<TestCommand as Command>::ER>,
+        )>,
+        <TestCommand as Command>::Error,
+    > {
+        let as_duration = cmds
+            .iter()
+            .fold(Duration::default(), |acc, c| acc + c.cmd().as_dur);
+        sleep(as_duration).await;
+        if cmds.iter().any(|c| c.cmd().as_should_fail) {
             return Err(ExecuteError("fail".to_owned()));
         }
-        self.after_sync_sender
-            .send((cmd.clone(), index))
-            .expect("failed to send after sync msg");
+        let total = cmds.len();
+        for (i, cmd) in cmds.iter().enumerate() {
+            let index = highest_index - (total - i - 1) as u64;
+            self.after_sync_sender
+                .send((cmd.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(),
+            highest_index.to_le_bytes().to_vec(),
         )];
-        if let TestCommandType::Put(v) = cmd.cmd_type {
-            debug!("cmd {:?}-{:?} revision is {}", cmd.cmd_type, cmd, revision);
-            let value = v.to_le_bytes().to_vec();
-            let keys = cmd
-                .keys
-                .iter()
-                .map(|k| k.to_le_bytes().to_vec())
-                .collect_vec();
-            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_le_bytes().to_vec(),
-                        )
-                    })),
+
+        let mut asrs = Vec::new();
+        for (i, c) in cmds.iter().enumerate() {
+            let cmd = c.cmd();
+            let index = highest_index - (total - i) as u64;
+            asrs.push((LogIndexResult(index), None));
+            if let TestCommandType::Put(v) = cmd.cmd_type {
+                let revision = self.revision.fetch_add(1, Ordering::Relaxed);
+                debug!("cmd {:?}-{:?} revision is {}", cmd.cmd_type, cmd, revision);
+                let value = v.to_le_bytes().to_vec();
+                let keys = cmd
+                    .keys
+                    .iter()
+                    .map(|k| k.to_le_bytes().to_vec())
+                    .collect_vec();
+                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_le_bytes().to_vec(),
+                            )
+                        })),
+                );
+            }
+            debug!(
+                "{} after sync cmd({:?} - {:?}), index: {index}",
+                self.server_name, cmd.cmd_type, cmd
             );
-            self.store
-                .write_multi(wr_ops, true)
-                .map_err(|e| ExecuteError(e.to_string()))?;
         }
-        debug!(
-            "{} after sync cmd({:?} - {:?}), index: {index}",
-            self.server_name, cmd.cmd_type, cmd
-        );
-        Ok(index.into())
+
+        self.store
+            .write_multi(wr_ops, true)
+            .map_err(|e| ExecuteError(e.to_string()))?;
+        Ok(asrs)
     }
 
     fn set_last_applied(&self, index: LogIndex) -> Result<(), <TestCommand as Command>::Error> {