Merge remote-tracking branch 'upstream/main' into feature_cluster

Author: zhyass

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

@@ -73,7 +73,8 @@ where SM: StateMachineApi + 'static
     pub(crate) cmd_ctx: CmdContext,
 
     /// The changes have been made by the applying one log entry
-    changes: Vec<Change<Vec<u8>, String>>,
+    /// `(key, prev, result)`.
+    changes: Vec<(String, Option<SeqV>, Option<SeqV>)>,
 }
 
 impl<'a, SM> Applier<'a, SM>
@@ -586,8 +587,7 @@ where SM: StateMachineApi + 'static
             return;
         }
 
-        self.changes
-            .push(Change::new(prev, result).with_id(key.to_string()))
+        self.changes.push((key.to_string(), prev, result))
     }
 
     /// Retrieve the proposing time from a raft-log.

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

@@ -16,7 +16,6 @@ use std::fmt::Debug;
 
 use databend_common_meta_types::protobuf::WatchResponse;
 use databend_common_meta_types::sys_data::SysData;
-use databend_common_meta_types::Change;
 use databend_common_meta_types::SeqV;
 use tokio::sync::mpsc;
 use tonic::Status;
@@ -27,7 +26,7 @@ use crate::state_machine::ExpireKey;
 
 /// Send a key-value change event to subscribers.
 pub trait SMEventSender: Debug + Sync + Send {
-    fn send(&self, change: Change<Vec<u8>, String>);
+    fn send(&self, change: (String, Option<SeqV>, Option<SeqV>));
 
     /// Inform to send all items in `strm` to `tx`.
     ///
@@ -39,21 +38,51 @@ pub trait SMEventSender: Debug + Sync + Send {
     );
 }
 
-/// The API a state machine implements
+/// The API a state machine implements.
+///
+/// The state machine is responsible for managing the application's persistent state,
+/// including application kv data and expired key data.
 pub trait StateMachineApi: Send + Sync {
     type Map: MapApi<String> + MapApi<ExpireKey> + 'static;
 
+    /// Returns the current expire key cursor position.
+    ///
+    /// The expiry key cursor marks a boundary in the key space:
+    /// - All keys before this cursor (exclusive) have already been processed and deleted
+    /// - This cursor position is used to track progress when incrementally cleaning up expired keys
     fn get_expire_cursor(&self) -> ExpireKey;
 
+    /// Updates the expiry key cursor position.
+    ///
+    /// This method is called after a batch of expired keys have been processed and deleted.
+    /// The new cursor position indicates that all keys before it (exclusive) have been
+    /// successfully cleaned up.
     fn set_expire_cursor(&mut self, cursor: ExpireKey);
 
-    /// Return a reference to the map that stores app data.
+    /// Returns a reference to the map that stores application data.
+    ///
+    /// This method provides read-only access to the underlying key-value store
+    /// that contains the application's persistent state, including application kv data and expired key data.
     fn map_ref(&self) -> &Self::Map;
 
-    /// Return a mutable reference to the map that stores app data.
+    /// Returns a mutable reference to the map that stores application data.
+    ///
+    /// This method provides read-write access to the underlying key-value store
+    /// that contains the application's persistent state, including application kv data and expired key data.
+    /// Changes made through this reference will be persisted according to the state machine's replication
+    /// protocol.
     fn map_mut(&mut self) -> &mut Self::Map;
 
+    /// Returns a mutable reference to the system data.
+    ///
+    /// This method provides read-write access to the system data, which includes
+    /// metadata about the state machine and its configuration.
     fn sys_data_mut(&mut self) -> &mut SysData;
 
+    /// Returns an optional reference to the event sender.
+    ///
+    /// This method returns an event sender that can be used to send state change events to subscribers.
+    ///
+    /// The implementation could just return `None` if the state machine does not support subscribing.
     fn event_sender(&self) -> Option<&dyn SMEventSender>;
 }

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

@@ -78,7 +78,7 @@ use crate::version::from_digit_ver;
 use crate::version::to_digit_ver;
 use crate::version::METASRV_SEMVER;
 use crate::version::MIN_METACLI_SEMVER;
-use crate::watcher::WatchStream;
+use crate::watcher::watch_stream::WatchStream;
 
 pub struct MetaServiceImpl {
     token: GrpcToken,
@@ -396,7 +396,7 @@ impl MetaService for MetaServiceImpl {
         let mn = &self.meta_node;
 
         let sender = mn.add_watcher(watch, tx.clone()).await?;
-        let stream = WatchStream::new(rx, sender, mn.subscriber_handle.clone());
+        let stream = WatchStream::new(rx, sender, mn.dispatcher_handle.clone());
 
         if flush {
             let sm = mn.raft_store.state_machine.clone();

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

@@ -94,9 +94,9 @@ use crate::request_handling::Forwarder;
 use crate::request_handling::Handler;
 use crate::store::RaftStore;
 use crate::version::METASRV_COMMIT_VERSION;
-use crate::watcher::EventSubscriber;
-use crate::watcher::StreamSender;
-use crate::watcher::SubscriberHandle;
+use crate::watcher::dispatch::Dispatcher;
+use crate::watcher::dispatch::DispatcherHandle;
+use crate::watcher::watch_stream::WatchStreamSender;
 
 pub type LogStore = RaftStore;
 pub type SMStore = RaftStore;
@@ -107,7 +107,7 @@ pub type MetaRaft = Raft<TypeConfig>;
 /// MetaNode is the container of metadata related components and threads, such as storage, the raft node and a raft-state monitor.
 pub struct MetaNode {
     pub raft_store: RaftStore,
-    pub subscriber_handle: SubscriberHandle,
+    pub dispatcher_handle: DispatcherHandle,
     pub raft: MetaRaft,
     pub running_tx: watch::Sender<()>,
     pub running_rx: watch::Receiver<()>,
@@ -159,15 +159,15 @@ impl MetaNodeBuilder {
 
         let (tx, rx) = watch::channel::<()>(());
 
-        let handle = EventSubscriber::spawn();
+        let handle = Dispatcher::spawn();
 
         sto.get_state_machine()
             .await
             .set_event_sender(Box::new(handle.clone()));
 
         let meta_node = Arc::new(MetaNode {
             raft_store: sto.clone(),
-            subscriber_handle: handle,
+            dispatcher_handle: handle,
             raft: raft.clone(),
             running_tx: tx,
             running_rx: rx,
@@ -1148,12 +1148,12 @@ impl MetaNode {
         &self,
         request: WatchRequest,
         tx: mpsc::Sender<Result<WatchResponse, Status>>,
-    ) -> Result<Arc<StreamSender>, Status> {
+    ) -> Result<Arc<WatchStreamSender>, Status> {
         let stream_sender = self
-            .subscriber_handle
-            .request_blocking(|d: &mut EventSubscriber| d.add_watcher(request, tx))
+            .dispatcher_handle
+            .request_blocking(|d: &mut Dispatcher| d.add_watcher(request, tx))
             .await
-            .map_err(|_e| Status::internal("EventSubscriber closed"))?
+            .map_err(|_e| Status::internal("watch-event-Dispatcher closed"))?
             .map_err(Status::invalid_argument)?;
 
         Ok(stream_sender)

src/meta/service/src/watcher/command.rs renamed to src/meta/service/src/watcher/dispatch/command.rs

@@ -12,26 +12,40 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use databend_common_meta_types::Change;
+use databend_common_meta_types::SeqV;
 use futures::future::BoxFuture;
 
-use crate::watcher::subscriber::EventSubscriber;
+use crate::watcher::dispatch::Dispatcher;
 
-/// An event sent to EventDispatcher.
+/// A command sent to [`Dispatcher`].
 #[allow(clippy::type_complexity)]
-pub(crate) enum Command {
-    /// Submit a kv change event to dispatcher
-    KVChange(Change<Vec<u8>, String>),
+pub enum Command {
+    /// Submit a key-value update event to dispatcher.
+    Update(Update),
 
-    /// Send a fn to [`EventSubscriber`] to run it.
+    /// Send a fn to [`Dispatcher`] to run it.
     ///
     /// The function will be called with a mutable reference to the dispatcher.
     Request {
-        req: Box<dyn FnOnce(&mut EventSubscriber) + Send + 'static>,
+        req: Box<dyn FnOnce(&mut Dispatcher) + Send + 'static>,
     },
 
-    /// Send a fn to [`EventSubscriber`] to run it asynchronously.
+    /// Send a fn to [`Dispatcher`] to run it asynchronously.
     RequestAsync {
-        req: Box<dyn FnOnce(&mut EventSubscriber) -> BoxFuture<'static, ()> + Send + 'static>,
+        req: Box<dyn FnOnce(&mut Dispatcher) -> BoxFuture<'static, ()> + Send + 'static>,
     },
 }
+
+/// An update event for a key.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Update {
+    pub key: String,
+    pub before: Option<SeqV>,
+    pub after: Option<SeqV>,
+}
+
+impl Update {
+    pub fn new(key: String, before: Option<SeqV>, after: Option<SeqV>) -> Self {
+        Self { key, before, after }
+    }
+}