refactor(meta-service): simplify StateMachineApi, remove SMEventSender trait (#18368)

Author: drmingdrmer

src/meta/raft-store/src/applier/mod.rs

@@ -130,11 +130,9 @@ where SM: StateMachineApi + 'static
         };
 
         // Send queued change events to subscriber
-        if let Some(sender) = self.sm.event_sender() {
-            for event in self.changes.drain(..) {
-                debug!("send to EventSender: {:?}", event);
-                sender.send(event);
-            }
+        for event in self.changes.drain(..) {
+            debug!("send to EventSender: {:?}", event);
+            self.sm.on_change_applied(event);
         }
 
         Ok(applied_state)

src/meta/raft-store/src/sm_v003/sm_v003.rs

@@ -12,14 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::Debug;
+use std::fmt;
+use std::fmt::Formatter;
 use std::io;
 
 use databend_common_meta_types::raft_types::Entry;
 use databend_common_meta_types::raft_types::StorageError;
 use databend_common_meta_types::snapshot_db::DB;
 use databend_common_meta_types::sys_data::SysData;
 use databend_common_meta_types::AppliedState;
+use databend_common_meta_types::SeqV;
 use log::info;
 use openraft::entry::RaftEntry;
 
@@ -29,18 +31,32 @@ use crate::leveled_store::leveled_map::LeveledMap;
 use crate::leveled_store::sys_data_api::SysDataApiRO;
 use crate::sm_v003::sm_v003_kv_api::SMV003KVApi;
 use crate::state_machine::ExpireKey;
-use crate::state_machine_api::SMEventSender;
 use crate::state_machine_api::StateMachineApi;
 
-#[derive(Debug, Default)]
+type OnChange = Box<dyn Fn((String, Option<SeqV>, Option<SeqV>)) + Send + Sync>;
+
+#[derive(Default)]
 pub struct SMV003 {
     levels: LeveledMap,
 
     /// The expiration key since which for next clean.
     expire_cursor: ExpireKey,
 
-    /// subscriber of state machine data
-    pub(crate) subscriber: Option<Box<dyn SMEventSender>>,
+    /// Callback when a change is applied to state machine
+    pub(crate) on_change_applied: Option<OnChange>,
+}
+
+impl fmt::Debug for SMV003 {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SMV003")
+            .field("levels", &self.levels)
+            .field("expire_cursor", &self.expire_cursor)
+            .field(
+                "on_change_applied",
+                &self.on_change_applied.as_ref().map(|_x| "is_set"),
+            )
+            .finish()
+    }
 }
 
 impl StateMachineApi for SMV003 {
@@ -68,8 +84,12 @@ impl StateMachineApi for SMV003 {
         self.levels.sys_data_mut()
     }
 
-    fn event_sender(&self) -> Option<&dyn SMEventSender> {
-        self.subscriber.as_ref().map(|x| x.as_ref())
+    fn on_change_applied(&mut self, change: (String, Option<SeqV>, Option<SeqV>)) {
+        let Some(on_change_applied) = &self.on_change_applied else {
+            // No subscribers, do nothing.
+            return;
+        };
+        (*on_change_applied)(change);
     }
 }
 
@@ -170,8 +190,8 @@ impl SMV003 {
         self.map_mut()
     }
 
-    pub fn set_event_sender(&mut self, subscriber: Box<dyn SMEventSender>) {
-        self.subscriber = Some(subscriber);
+    pub fn set_on_change_applied(&mut self, on_change_applied: OnChange) {
+        self.on_change_applied = Some(on_change_applied);
     }
 
     pub fn freeze_writable(&mut self) {

src/meta/raft-store/src/state_machine_api.rs

@@ -12,37 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::Debug;
-use std::sync::Arc;
-
 use databend_common_meta_types::sys_data::SysData;
 use databend_common_meta_types::SeqV;
-use futures::future::BoxFuture;
 use map_api::map_api::MapApi;
 
 use crate::state_machine::ExpireKey;
 use crate::state_machine::UserKey;
 
-/// Send a key-value change event to subscribers.
-pub trait SMEventSender: Debug + Sync + Send {
-    fn send(&self, change: (String, Option<SeqV>, Option<SeqV>));
-
-    /// Send a future to the worker to let it run it in serialized order.
-    fn send_future(&self, fut: BoxFuture<'static, ()>);
-}
-
-impl<T> SMEventSender for Arc<T>
-where T: SMEventSender
-{
-    fn send(&self, change: (String, Option<SeqV>, Option<SeqV>)) {
-        self.as_ref().send(change);
-    }
-
-    fn send_future(&self, fut: BoxFuture<'static, ()>) {
-        self.as_ref().send_future(fut);
-    }
-}
-
 /// The API a state machine implements.
 ///
 /// The state machine is responsible for managing the application's persistent state,
@@ -80,10 +56,17 @@ pub trait StateMachineApi: Send + Sync {
     /// metadata about the state machine and its configuration.
     fn sys_data_mut(&mut self) -> &mut SysData;
 
-    /// Returns an optional reference to the event sender.
+    /// Notify subscribers of a key-value change applied to the state machine.
+    ///
+    /// Called after a change is committed, but before it is guaranteed persisted.
+    /// The change may be replayed on server restart.
     ///
-    /// This method returns an event sender that can be used to send state change events to subscribers.
+    /// - `change`: (`String`, `Option<SeqV>`, `Option<SeqV>`)
+    ///   - key: user application key
+    ///   - old: previous value (`None` if new key)
+    ///   - new: new value (`None` if deleted)
     ///
-    /// The implementation could just return `None` if the state machine does not support subscribing.
-    fn event_sender(&self) -> Option<&dyn SMEventSender>;
+    /// Called for every successful create, update, or delete.
+    /// Implementations without subscribers may leave this empty.
+    fn on_change_applied(&mut self, change: (String, Option<SeqV>, Option<SeqV>));
 }

src/meta/raft-store/src/state_machine_api_ext.rs

@@ -122,14 +122,10 @@ pub trait StateMachineApiExt: StateMachineApi {
 
         let strm = if let Some(right) = prefix_right_bound(&p) {
             self.user_map()
-                .as_user_map()
                 .range(UserKey::new(&p)..UserKey::new(right))
                 .await?
         } else {
-            self.user_map()
-                .as_user_map()
-                .range(UserKey::new(&p)..)
-                .await?
+            self.user_map().range(UserKey::new(&p)..).await?
         };
 
         let strm = strm
@@ -218,7 +214,7 @@ pub trait StateMachineApiExt: StateMachineApi {
     ///
     /// It does not check expiration of the returned entry.
     async fn get_maybe_expired_kv(&self, key: &String) -> Result<Option<SeqV>, io::Error> {
-        let got = self.user_map().as_user_map().get(key.as_ref()).await?;
+        let got = self.user_map().get(key.as_ref()).await?;
         let seqv = Into::<Option<SeqV>>::into(got);
         Ok(seqv)
     }

src/meta/service/src/api/grpc/grpc_service.rs

@@ -29,7 +29,6 @@ use databend_common_meta_client::MetaGrpcReadReq;
 use databend_common_meta_client::MetaGrpcReq;
 use databend_common_meta_kvapi::kvapi::KVApi;
 use databend_common_meta_raft_store::state_machine::UserKey;
-use databend_common_meta_raft_store::state_machine_api::SMEventSender;
 use databend_common_meta_raft_store::state_machine_api_ext::StateMachineApiExt;
 use databend_common_meta_types::protobuf as pb;
 use databend_common_meta_types::protobuf::meta_service_server::MetaService;
@@ -74,6 +73,7 @@ use tonic::Request;
 use tonic::Response;
 use tonic::Status;
 use tonic::Streaming;
+use watcher::dispatch::Command;
 use watcher::key_range::build_key_range;
 use watcher::util::new_initialization_sink;
 use watcher::util::try_forward;
@@ -518,7 +518,7 @@ impl MetaService for MetaServiceImpl {
                     sender_str
                 );
 
-                mn.dispatcher_handle.send_future(fu);
+                mn.dispatcher_handle.send_command(Command::Future(fu));
             }
 
             stream

src/meta/service/src/meta_service/meta_node.rs

@@ -174,9 +174,14 @@ impl MetaNodeBuilder {
         let handle = DispatcherHandle::new(handle, node_id);
         let handle = Arc::new(handle);
 
+        let on_change_applied = {
+            let h = handle.clone();
+            move |change| h.send_change(change)
+        };
+
         sto.get_state_machine()
             .await
-            .set_event_sender(Box::new(handle.clone()));
+            .set_on_change_applied(Box::new(on_change_applied));
 
         let meta_node = Arc::new(MetaNode {
             raft_store: sto.clone(),

src/meta/service/src/meta_service/watcher.rs

@@ -19,13 +19,10 @@ use std::future::Future;
 use std::io::Error;
 use std::ops::Deref;
 
-use databend_common_meta_raft_store::state_machine_api::SMEventSender;
 use databend_common_meta_types::protobuf::WatchResponse;
 use databend_common_meta_types::SeqV;
-use futures::future::BoxFuture;
 use log::debug;
 use tonic::Status;
-use watcher::dispatch::Command;
 use watcher::dispatch::DispatcherHandle as GenericDispatcherHandle;
 use watcher::type_config::KVChange;
 use watcher::type_config::KeyOf;
@@ -109,13 +106,3 @@ impl DispatcherHandle {
         h
     }
 }
-
-impl SMEventSender for DispatcherHandle {
-    fn send(&self, change: KVChange<WatchTypes>) {
-        self.send_change(change);
-    }
-
-    fn send_future(&self, fut: BoxFuture<'static, ()>) {
-        self.send_command(Command::Future(fut));
-    }
-}