Support post_exec on Raftstore Proxy

.github/workflows/pr-ci.yml

@@ -72,5 +72,7 @@ jobs:
           cargo test --package tests --test failpoints cases::test_merge
           cargo test --package tests --test failpoints cases::test_import_service
           cargo test --package tests --test failpoints cases::test_proxy_replica_read
+          # tests based on new-mock-engine-store, with compaction for new proxy
           cargo test --features compat_new_proxy --package tests --test proxy normal
+          # tests based on new-mock-engine-store, for some tests not available for new proxy
           cargo test --package tests --test proxy proxy

components/cdc/src/delegate.rs

@@ -446,6 +446,7 @@ impl Delegate {
         for cmd in batch.into_iter(self.region_id) {
             let Cmd {
                 index,
+                term: _,
                 mut request,
                 mut response,
             } = cmd;

components/raftstore/src/coprocessor/dispatcher.rs

@@ -439,7 +439,38 @@ impl<E: KvEngine> CoprocessorHost<E> {
             let admin = cmd.get_admin_request();
             for observer in &self.registry.admin_observers {
                 let observer = observer.observer.inner();
-                if observer.pre_exec_admin(&mut ctx, admin, index, term) {
+                if observer.pre_exec_admin(&mut ctx, admin) {
+                    return true;
+                }
+            }
+            false
+        }
+    }
+
+    /// `post_exec` should be called immediately after we executed one raft command.
+    /// It notifies observers side effects of this command before execution of the next command,
+    /// including req/resp, apply state, modified region state, etc.
+    /// Return true observers think a persistence is necessary.
+    pub fn post_exec(
+        &self,
+        region: &Region,
+        cmd: &Cmd,
+        apply_state: &RaftApplyState,
+        region_state: &RegionState,
+    ) -> bool {
+        let mut ctx = ObserverContext::new(region);
+        if !cmd.response.has_admin_response() {
+            for observer in &self.registry.query_observers {
+                let observer = observer.observer.inner();
+                if observer.post_exec_query(&mut ctx, cmd, apply_state, region_state) {
+                    return true;
+                }
+            }
+            false
+        } else {
+            for observer in &self.registry.admin_observers {
+                let observer = observer.observer.inner();
+                if observer.post_exec_admin(&mut ctx, cmd, apply_state, region_state) {
                     return true;
                 }
             }
@@ -801,7 +832,7 @@ mod tests {
         assert_all!([&ob.called], &[3]);
         let mut admin_resp = RaftCmdResponse::default();
         admin_resp.set_admin_response(AdminResponse::default());
-        host.post_apply(&region, &Cmd::new(0, admin_req, admin_resp));
+        host.post_apply(&region, &Cmd::new(0, 0, admin_req, admin_resp));
         assert_all!([&ob.called], &[6]);
 
         let mut query_req = RaftCmdRequest::default();
@@ -811,7 +842,7 @@ mod tests {
         host.pre_apply(&region, &query_req);
         assert_all!([&ob.called], &[15]);
         let query_resp = RaftCmdResponse::default();
-        host.post_apply(&region, &Cmd::new(0, query_req, query_resp));
+        host.post_apply(&region, &Cmd::new(0, 0, query_req, query_resp));
         assert_all!([&ob.called], &[21]);
 
         host.on_role_change(&region, RoleChange::new(StateRole::Leader));
@@ -893,7 +924,7 @@ mod tests {
             host.pre_apply(&region, &req);
             assert_all!([&ob1.called, &ob2.called], &[0, base_score * 2 + 3]);
 
-            host.post_apply(&region, &Cmd::new(0, req.clone(), resp.clone()));
+            host.post_apply(&region, &Cmd::new(0, 0, req.clone(), resp.clone()));
             assert_all!([&ob1.called, &ob2.called], &[0, base_score * 3 + 6]);
 
             set_all!(&[&ob2.bypass], false);

components/raftstore/src/coprocessor/mod.rs

@@ -14,6 +14,7 @@ use kvproto::{
     metapb::Region,
     pdpb::CheckPolicy,
     raft_cmdpb::{AdminRequest, AdminResponse, RaftCmdRequest, RaftCmdResponse, Request},
+    raft_serverpb::RaftApplyState,
 };
 use raft::{eraftpb, StateRole};
 
@@ -74,6 +75,12 @@ impl<'a> ObserverContext<'a> {
     }
 }
 
+pub struct RegionState {
+    pub peer_id: u64,
+    pub pending_remove: bool,
+    pub modified_region: Option<Region>,
+}
+
 pub trait AdminObserver: Coprocessor {
     /// Hook to call before proposing admin request.
     fn pre_propose_admin(&self, _: &mut ObserverContext<'_>, _: &mut AdminRequest) -> Result<()> {
@@ -87,6 +94,18 @@ pub trait AdminObserver: Coprocessor {
     /// For now, the `region` in `ObserverContext` is an empty region.
     fn post_apply_admin(&self, _: &mut ObserverContext<'_>, _: &AdminResponse) {}
 
+    /// Hook to call immediately after exec command
+    /// Will be a special persistence after this exec if a observer returns true.
+    fn post_exec_admin(
+        &self,
+        _: &mut ObserverContext<'_>,
+        _: &Cmd,
+        _: &RaftApplyState,
+        _: &RegionState,
+    ) -> bool {
+        false
+    }
+
     /// Hook before exec admin request, returns whether we should skip this
     /// admin.
     fn pre_exec_admin(
@@ -118,6 +137,18 @@ pub trait QueryObserver: Coprocessor {
     /// For now, the `region` in `ObserverContext` is an empty region.
     fn post_apply_query(&self, _: &mut ObserverContext<'_>, _: &Cmd) {}
 
+    /// Hook to call immediately after exec command.
+    /// Will be a special persistence after this exec if a observer returns true.
+    fn post_exec_query(
+        &self,
+        _: &mut ObserverContext<'_>,
+        _: &Cmd,
+        _: &RaftApplyState,
+        _: &RegionState,
+    ) -> bool {
+        false
+    }
+
     /// Hook before exec write request, returns whether we should skip this
     /// write.
     fn pre_exec_query(&self, _: &mut ObserverContext<'_>, _: &[Request], _: u64, _: u64) -> bool {
@@ -241,14 +272,16 @@ pub trait RegionChangeObserver: Coprocessor {
 #[derive(Clone, Debug, Default)]
 pub struct Cmd {
     pub index: u64,
+    pub term: u64,
     pub request: RaftCmdRequest,
     pub response: RaftCmdResponse,
 }
 
 impl Cmd {
-    pub fn new(index: u64, request: RaftCmdRequest, response: RaftCmdResponse) -> Cmd {
+    pub fn new(index: u64, term: u64, request: RaftCmdRequest, response: RaftCmdResponse) -> Cmd {
         Cmd {
             index,
+            term,
             request,
             response,
         }