change: Add on_commit on_complete to Responder trait

Add `ProgressResponder` and use it as default `Responder` that receives
`on_commit` and `on_complete` events.

- `on_commit` is called when a log entry is committed;
- `on_complete` is called when the log entry is applied to state machine
  or an error occur.

Replace OneshotResponder with ProgressResponder throughout the codebase to
provide both commit and completion(apply or error) notifications by default.

- Fix: #1460

Upgrade tip:

When not using application defined `Responder`, nothing todo.

Otherwise, update the implementation of `Responder` by adding
`on_commit` if needed, and replacing `send` with `on_complete`.

Author: drmingdrmer

openraft/src/core/raft_core.rs

@@ -463,19 +463,19 @@ where
         &mut self,
         changes: ChangeMembers<C>,
         retain: bool,
-        tx: OneshotResponder<C, ClientWriteResult<C>>,
+        tx: ProgressResponder<C, ClientWriteResult<C>>,
     ) {
         let res = self.engine.state.membership_state.change_handler().apply(changes, retain);
         let new_membership = match res {
             Ok(x) => x,
             Err(e) => {
-                tx.send(Err(ClientWriteError::ChangeMembershipError(e)));
+                tx.on_complete(Err(ClientWriteError::ChangeMembershipError(e)));
                 return;
             }
         };
 
         let ent = C::Entry::new_membership(LogIdOf::<C>::default(), new_membership);
-        self.write_entry(ent, Some(CoreResponder::Oneshot(tx)));
+        self.write_entry(ent, Some(CoreResponder::Progress(tx)));
     }
 
     /// Write a log entry to the cluster through raft protocol.
@@ -487,7 +487,7 @@ where
     /// The calling side may not receive a result from `resp_tx`, if raft is shut down.
     ///
     /// The responder `resp_tx` is either Responder type of
-    /// [`RaftTypeConfig::Responder`] (application-defined) or [`OneshotResponder`]
+    /// [`RaftTypeConfig::Responder`] (application-defined) or [`ProgressResponder`]
     /// (general-purpose); the former is for application-defined entries like user data, the
     /// latter is for membership configuration changes.
     #[tracing::instrument(level = "debug", skip_all, fields(id = display(&self.id)))]
@@ -502,7 +502,7 @@ where
         if let Some(to) = lh.leader.get_transfer_to() {
             if let Some(tx) = tx {
                 let err = lh.state.new_forward_to_leader(to.clone());
-                tx.send(Err(ClientWriteError::ForwardToLeader(err)));
+                tx.on_complete(Err(ClientWriteError::ForwardToLeader(err)));
             }
             return;
         }
@@ -818,6 +818,11 @@ where
         let entry_count = last.index() + 1 - first.index();
         self.runtime_stats.apply_batch.record(entry_count);
 
+        for (index, responder) in responders.iter_mut() {
+            let log_id = self.engine.state.get_log_id(*index).unwrap();
+            responder.on_commit(log_id);
+        }
+
         let cmd = sm::Command::apply(first, last.clone(), responders);
         self.sm_handle.send(cmd).map_err(|e| StorageError::apply(last, AnyError::error(e)))?;
 
@@ -1829,7 +1834,7 @@ where
                     #[allow(clippy::let_underscore_future)]
                     let _ = C::spawn(async move {
                         for (log_index, tx) in removed.into_iter() {
-                            tx.send(Err(ClientWriteError::ForwardToLeader(ForwardToLeader {
+                            tx.on_complete(Err(ClientWriteError::ForwardToLeader(ForwardToLeader {
                                 leader_id: leader_id.clone(),
                                 leader_node: leader_node.clone(),
                             })));

openraft/src/core/raft_msg/mod.rs

@@ -93,7 +93,7 @@ where C: RaftTypeConfig
         /// config will be converted into learners, otherwise they will be removed.
         retain: bool,
 
-        tx: OneshotResponder<C, ClientWriteResult<C>>,
+        tx: ProgressResponder<C, ClientWriteResult<C>>,
     },
 
     ExternalCoreRequest {

openraft/src/engine/engine_impl.rs

@@ -262,7 +262,7 @@ where C: RaftTypeConfig
         tx: Option<R>,
     ) -> Option<(LeaderHandler<'_, C>, Option<R>)>
     where
-        R: Responder<ClientWriteResult<C>>,
+        R: Responder<C, ClientWriteResult<C>>,
     {
         let res = self.leader_handler();
         let forward_err = match res {
@@ -274,7 +274,7 @@ where C: RaftTypeConfig
         };
 
         if let Some(tx) = tx {
-            tx.send(Err(forward_err.into()));
+            tx.on_complete(Err(forward_err.into()));
         }
 
         None

openraft/src/impls/mod.rs

@@ -27,6 +27,7 @@ pub use crate::entry::Entry;
 pub use crate::node::BasicNode;
 pub use crate::node::EmptyNode;
 pub use crate::raft::responder::impls::OneshotResponder;
+pub use crate::raft::responder::impls::ProgressResponder;
 #[cfg(feature = "tokio-rt")]
 pub use crate::type_config::async_runtime::tokio_impls::TokioRuntime;
 

openraft/src/raft/api/app.rs

@@ -50,12 +50,11 @@ where C: RaftTypeConfig
         app_data: C::D,
         // TODO: ClientWriteError can only be ForwardToLeader Error
     ) -> Result<Result<ClientWriteResponse<C>, ClientWriteError<C>>, Fatal<C>> {
-        let (tx, rx) = C::oneshot();
-        let responder = OneshotResponder::new(tx);
+        let (responder, _commit_rx, complete_rx) = ProgressResponder::new();
 
-        self.do_client_write_ff(app_data, Some(CoreResponder::Oneshot(responder))).await?;
+        self.do_client_write_ff(app_data, Some(CoreResponder::Progress(responder))).await?;
 
-        let res: ClientWriteResult<C> = self.inner.recv_msg(rx).await?;
+        let res: ClientWriteResult<C> = self.inner.recv_msg(complete_rx).await?;
 
         Ok(res)
     }

openraft/src/raft/api/management.rs

@@ -71,7 +71,7 @@ where C: RaftTypeConfig
             "change_membership: start to commit joint config"
         );
 
-        let (tx, rx) = oneshot_channel::<C, _>();
+        let (tx, rx) = new_responder_pair::<C, _>();
 
         // res is error if membership cannot be changed.
         // If no error, it will enter a joint state
@@ -106,7 +106,7 @@ where C: RaftTypeConfig
         tracing::debug!("committed a joint config: {} {:?}", log_id, joint);
         tracing::debug!("the second step is to change to uniform config: {:?}", changes);
 
-        let (tx, rx) = oneshot_channel::<C, _>();
+        let (tx, rx) = new_responder_pair::<C, _>();
 
         // The second step, send a NOOP change to flatten the joint config.
         let changes = ChangeMembers::AddVoterIds(Default::default());
@@ -132,7 +132,7 @@ where C: RaftTypeConfig
         node: C::Node,
         blocking: bool,
     ) -> Result<ClientWriteResult<C>, Fatal<C>> {
-        let (tx, rx) = oneshot_channel::<C, _>();
+        let (tx, rx) = new_responder_pair::<C, _>();
 
         let msg = RaftMsg::ChangeMembership {
             changes: ChangeMembers::AddNodes(btreemap! {id.clone()=>node}),
@@ -229,14 +229,12 @@ where C: RaftTypeConfig
     }
 }
 
-fn oneshot_channel<C, T>() -> (OneshotResponder<C, T>, OneshotReceiverOf<C, T>)
+fn new_responder_pair<C, T>() -> (ProgressResponder<C, T>, OneshotReceiverOf<C, T>)
 where
     C: RaftTypeConfig,
     T: OptionalSend,
 {
-    let (tx, rx) = C::oneshot();
+    let (tx, _commit_rx, complete_rx) = ProgressResponder::new();
 
-    let tx = OneshotResponder::new(tx);
-
-    (tx, rx)
+    (tx, complete_rx)
 }

openraft/src/raft/mod.rs

@@ -190,7 +190,7 @@ macro_rules! declare_raft_types {
                 (Vote           , , $crate::impls::Vote<Self>                    ),
                 (Entry          , , $crate::impls::Entry<Self>                   ),
                 (SnapshotData   , , std::io::Cursor<Vec<u8>>                     ),
-                (Responder<T>   , , $crate::impls::OneshotResponder<Self, T> where T: $crate::OptionalSend + 'static     ),
+                (Responder<T>   , , $crate::impls::ProgressResponder<Self, T> where T: $crate::OptionalSend + 'static     ),
                 (AsyncRuntime   , , $crate::impls::TokioRuntime                  ),
             );
 

openraft/src/raft/responder/core_responder.rs

@@ -1,3 +1,4 @@
+use crate::LogId;
 use crate::RaftTypeConfig;
 use crate::impls::OneshotResponder;
 use crate::raft::ClientWriteResult;
@@ -7,21 +8,28 @@ use crate::type_config::alias::WriteResponderOf;
 /// The responder used in RaftCore.
 ///
 /// RaftCore use this responder to send response to the caller.
-/// It is either an oneshot responder or a user-defined responder.
+/// It is either a progress responder or a user-defined responder.
 pub(crate) enum CoreResponder<C>
 where C: RaftTypeConfig
 {
-    Oneshot(OneshotResponder<C, ClientWriteResult<C>>),
+    Progress(ProgressResponder<C, ClientWriteResult<C>>),
     UserDefined(WriteResponderOf<C>),
 }
 
-impl<C> Responder<ClientWriteResult<C>> for CoreResponder<C>
+impl<C> Responder<C, ClientWriteResult<C>> for CoreResponder<C>
 where C: RaftTypeConfig
 {
-    fn send(self, res: ClientWriteResult<C>) {
+    fn on_commit(&mut self, log_id: LogId<C>) {
         match self {
-            Self::Oneshot(responder) => responder.send(res),
-            Self::UserDefined(responder) => responder.send(res),
+            Self::Progress(responder) => responder.on_commit(log_id),
+            Self::UserDefined(responder) => responder.on_commit(log_id),
+        }
+    }
+
+    fn on_complete(self, res: ClientWriteResult<C>) {
+        match self {
+            Self::Progress(responder) => responder.on_complete(res),
+            Self::UserDefined(responder) => responder.on_complete(res),
         }
     }
 }

openraft/src/raft/responder/impls/mod.rs

@@ -0,0 +1,5 @@
+mod oneshot_responder;
+mod progress_responder;
+
+pub use oneshot_responder::OneshotResponder;
+pub use progress_responder::ProgressResponder;

openraft/src/raft/responder/impls.rs renamed to openraft/src/raft/responder/impls/oneshot_responder.rs

@@ -47,12 +47,12 @@ where
     }
 }
 
-impl<C, T> Responder<T> for OneshotResponder<C, T>
+impl<C, T> Responder<C, T> for OneshotResponder<C, T>
 where
     C: RaftTypeConfig,
     T: OptionalSend + 'static,
 {
-    fn send(self, res: T) {
+    fn on_complete(self, res: T) {
         let res = self.tx.send(res);
 
         if res.is_ok() {