refactor: remove read state from kv_server

refactor: disable fast path in etcd competible layer

WIP: fix rebase kv server
Signed-off-by: bsbds <[email protected]>

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

@@ -170,7 +170,7 @@ where
     async fn reset(&self, snapshot: Option<(Snapshot, LogIndex)>) -> Result<(), C::Error>;
 
     /// Trigger the barrier of the given trigger id (based on propose id) and log index.
-    fn trigger(&self, id: InflightId, index: LogIndex);
+    fn trigger(&self, id: InflightId);
 }
 
 /// Codec for encoding and decoding data into/from the Protobuf format

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

@@ -412,7 +412,7 @@ impl CommandExecutor<TestCommand> for TestCE {
         Ok(())
     }
 
-    fn trigger(&self, _id: InflightId, _index: LogIndex) {}
+    fn trigger(&self, _id: InflightId) {}
 }
 
 impl TestCE {

crates/curp/src/server/cmd_worker/conflict_checked_mpmc.rs

@@ -348,7 +348,7 @@ impl<C: Command, CE: CommandExecutor<C>> Filter<C, CE> {
                                     None
                                 }
                                 Err(err) => {
-                                    self.cmd_executor.trigger(entry.inflight_id(), entry.index);
+                                    self.cmd_executor.trigger(entry.inflight_id());
                                     Some(err)
                                 }
                             }

crates/curp/src/server/cmd_worker/mod.rs

@@ -137,7 +137,7 @@ async fn worker_exe<C: Command, CE: CommandExecutor<C>, RC: RoleChange>(
         | EntryData::SetNodeState(_, _, _) => true,
     };
     if !success {
-        ce.trigger(entry.inflight_id(), entry.index);
+        ce.trigger(entry.inflight_id());
     }
     success
 }
@@ -240,7 +240,7 @@ async fn worker_as<C: Command, CE: CommandExecutor<C>, RC: RoleChange>(
         }
         EntryData::Empty => true,
     };
-    ce.trigger(entry.inflight_id(), entry.index);
+    ce.trigger(entry.inflight_id());
     success
 }
 

crates/xline/src/server/command.rs

@@ -18,7 +18,6 @@ use xlineapi::{
     AlarmAction, AlarmRequest, AlarmType,
 };
 
-use super::barriers::IndexBarrier;
 use crate::{
     revision_number::RevisionNumberGenerator,
     rpc::{RequestBackend, RequestWrapper},
@@ -73,8 +72,6 @@ pub(crate) struct CommandExecutor {
     alarm_storage: Arc<AlarmStore>,
     /// persistent storage
     db: Arc<DB>,
-    /// Barrier for applied index
-    index_barrier: Arc<IndexBarrier>,
     /// Barrier for propose id
     id_barrier: Arc<IdBarrier<InflightId>>,
     /// Revision Number generator for KV request and Lease request
@@ -225,7 +222,6 @@ impl CommandExecutor {
         lease_storage: Arc<LeaseStore>,
         alarm_storage: Arc<AlarmStore>,
         db: Arc<DB>,
-        index_barrier: Arc<IndexBarrier>,
         id_barrier: Arc<IdBarrier<InflightId>>,
         general_rev: Arc<RevisionNumberGenerator>,
         auth_rev: Arc<RevisionNumberGenerator>,
@@ -240,7 +236,6 @@ impl CommandExecutor {
             lease_storage,
             alarm_storage,
             db,
-            index_barrier,
             id_barrier,
             general_rev,
             auth_rev,
@@ -406,9 +401,8 @@ impl CurpCommandExecutor<Command> for CommandExecutor {
         Ok(u64::from_le_bytes(buf))
     }
 
-    fn trigger(&self, id: InflightId, index: LogIndex) {
+    fn trigger(&self, id: InflightId) {
         self.id_barrier.trigger(&id);
-        self.index_barrier.trigger(index);
     }
 }