Merge pull request #9240 from drmingdrmer/30-fixup

refactor(meta/watcher): just use Change as event type

Author: BohuTANG

src/meta/api/src/schema_api_impl.rs

@@ -987,7 +987,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_put(&dbid_tbname, serialize_u64(table_id)?), /* (tenant, db_id, tb_name) -> tb_id */
                         txn_op_put(&tbid, serialize_struct(&req.table_meta)?), /* (tenant, db_id, tb_id) -> tb_meta */
@@ -1149,7 +1148,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_del(&dbid_tbname), // (db_id, tb_name) -> tb_id
                         txn_op_put(&tbid, serialize_struct(&tb_meta)?), /* (tenant, db_id, tb_id) -> tb_meta */
@@ -1307,7 +1305,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_put(&dbid_tbname, serialize_u64(table_id)?), /* (tenant, db_id, tb_name) -> tb_id */
                         // txn_op_put(&dbid_tbname_idlist, serialize_struct(&tb_id_list)?)?, // _fd_table_id_list/db_id/table_name -> tb_id_list
@@ -1498,7 +1495,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     txn_op_put(&newdbid_newtbname, serialize_u64(table_id)?), /* (db_id, new_tb_name) -> tb_id */
                     // Changing a table in a db has to update the seq of db_meta,
                     // to block the batch-delete-tables when deleting a db.
-                    // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                     txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                     txn_op_put(&dbid_tbname_idlist, serialize_struct(&tb_id_list)?), /* _fd_table_id_list/db_id/old_table_name -> tb_id_list */
                     txn_op_put(&new_dbid_tbname_idlist, serialize_struct(&new_tb_id_list)?), /* _fd_table_id_list/db_id/new_table_name -> tb_id_list */
@@ -1507,7 +1503,6 @@ impl<KV: KVApi> SchemaApi for KV {
 
                 if db_id != new_db_id {
                     then_ops.push(
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(
                             &DatabaseId { db_id: new_db_id },
                             serialize_struct(&new_db_meta)?,

src/meta/raft-store/src/state_machine/sm.rs

@@ -95,11 +95,11 @@ const TREE_STATE_MACHINE: &str = "state_machine";
 
 /// StateMachine subscriber trait
 pub trait StateMachineSubscriber: Debug + Sync + Send {
-    fn kv_changed(&self, key: &str, prev: Option<SeqV>, current: Option<SeqV>);
+    fn kv_changed(&self, change: Change<Vec<u8>, String>);
 }
 
 /// The state machine of the `MemStore`.
-/// It includes user data and two raft-related informations:
+/// It includes user data and two raft-related information:
 /// `last_applied_logs` and `client_serial_responses` to achieve idempotence.
 #[derive(Debug)]
 pub struct StateMachine {
@@ -342,9 +342,7 @@ impl StateMachine {
         // Send queued change events to subscriber
         if let Some(subscriber) = &self.subscriber {
             for event in changes {
-                // TODO: use Change as the event data type.
-                // safe unwrap
-                subscriber.kv_changed(&event.ident.unwrap(), event.prev, event.result);
+                subscriber.kv_changed(event);
             }
         }
 
@@ -638,17 +636,15 @@ impl StateMachine {
         if let Some(kv_pairs) = kv_pairs {
             if let Some(kv_pairs) = kv_pairs.get(delete_by_prefix) {
                 for (key, _seq) in kv_pairs.iter() {
-                    // TODO: return StorageError
-                    let ret = Self::txn_upsert_kv(txn_tree, &UpsertKV::delete(key), log_time_ms);
+                    let (expired, prev, res) =
+                        Self::txn_upsert_kv(txn_tree, &UpsertKV::delete(key), log_time_ms)?;
 
-                    if let Ok(ret) = ret {
-                        count += 1;
+                    count += 1;
 
-                        if ret.0.is_some() {
-                            txn_tree.push_change(key, ret.0.clone(), None);
-                        }
-                        txn_tree.push_change(key, ret.1.clone(), ret.2);
+                    if expired.is_some() {
+                        txn_tree.push_change(key, expired, None);
                     }
+                    txn_tree.push_change(key, prev, res);
                 }
             }
         }
@@ -1046,7 +1042,7 @@ impl StateMachine {
         sm_nodes.range_values(..)
     }
 
-    /// Expire an `SeqV` and returns:
+    /// Expire an `SeqV` and returns the value discarded by expiration and the unexpired value:
     /// - `(Some, None)` if it expires.
     /// - `(None, Some)` if it does not.
     /// - `(None, None)` if the input is None.
@@ -1064,32 +1060,6 @@ impl StateMachine {
             (None, None)
         }
     }
-
-    pub fn unexpired<V: Debug>(seq_value: SeqV<V>, log_time_ms: u64) -> Option<SeqV<V>> {
-        // Caveat: The cleanup must be consistent across raft nodes:
-        //         A conditional update, e.g. an upsert_kv() with MatchSeq::Eq(some_value),
-        //         must be applied with the same timestamp on every raft node.
-        //         Otherwise: node-1 could have applied a log with a ts that is smaller than value.expire_at,
-        //         while node-2 may fail to apply the same log if it use a greater ts > value.expire_at.
-        //
-        // Thus:
-        //
-        // 1. A raft log must have a field ts assigned by the leader. When applying, use this ts to
-        //    check against expire_at to decide whether to purge it.
-        // 2. A GET operation must not purge any expired entry. Since a GET is only applied to a node itself.
-        // 3. The background task can only be triggered by the raft leader, by submit a "clean expired" log.
-
-        // TODO(xp): background task to clean expired
-        // TODO(xp): maybe it needs a expiration queue for efficient cleaning up.
-
-        debug!("seq_value: {:?} log_time_ms: {}", seq_value, log_time_ms);
-
-        if seq_value.get_expire_at() < log_time_ms {
-            None
-        } else {
-            Some(seq_value)
-        }
-    }
 }
 
 /// Key space support

src/meta/raft-store/src/state_machine/sm_kv_api_impl.rs

@@ -75,13 +75,10 @@ impl KVApi for StateMachine {
         // TODO(xp) refine get(): a &str is enough for key
         let sv = self.kvs().get(&key.to_string())?;
         debug!("get_kv sv:{:?}", sv);
-        let seq_v = match sv {
-            None => return Ok(None),
-            Some(sv) => sv,
-        };
 
         let local_now_ms = SeqV::<()>::now_ms();
-        Ok(Self::unexpired(seq_v, local_now_ms))
+        let (_expired, res) = Self::expire_seq_v(sv, local_now_ms);
+        Ok(res)
     }
 
     async fn mget_kv(&self, keys: &[String]) -> Result<MGetKVReply, KVAppError> {
@@ -111,7 +108,7 @@ impl KVApi for StateMachine {
         let local_now_ms = SeqV::<()>::now_ms();
 
         // Convert expired to None
-        let x = x.map(|(k, v)| (k, Self::unexpired(v, local_now_ms)));
+        let x = x.map(|(k, v)| (k, Self::expire_seq_v(Some(v), local_now_ms).1));
         // Remove None
         let x = x.filter(|(_k, v)| v.is_some());
         // Extract from an Option

src/meta/service/src/meta_service/raftmeta.rs

@@ -329,20 +329,18 @@ impl MetaNode {
         Ok(())
     }
 
-    /// Open or create a metasrv node.
-    /// Optionally boot a single node cluster.
+    /// Open or create a meta node.
     /// 1. If `open` is `Some`, try to open an existent one.
     /// 2. If `create` is `Some`, try to create an one in non-voter mode.
     #[tracing::instrument(level = "debug", skip_all)]
-    pub async fn open_create_boot(
+    pub async fn open_create(
         config: &RaftConfig,
         open: Option<()>,
         create: Option<()>,
-        initialize_cluster: Option<Node>,
     ) -> Result<Arc<MetaNode>, MetaStartupError> {
         info!(
-            "open_create_boot, config: {:?}, open: {:?}, create: {:?}, initialize_cluster: {:?}",
-            config, open, create, initialize_cluster
+            "open_create_boot, config: {:?}, open: {:?}, create: {:?}",
+            config, open, create
         );
 
         let mut config = config.clone();
@@ -371,10 +369,21 @@ impl MetaNode {
 
         info!("MetaNode started: {:?}", config);
 
-        // init_cluster with advertise_host other than listen_host
-        if mn.is_opened() {
-            return Ok(mn);
-        }
+        Ok(mn)
+    }
+
+    /// Open or create a metasrv node.
+    /// Optionally boot a single node cluster.
+    /// 1. If `open` is `Some`, try to open an existent one.
+    /// 2. If `create` is `Some`, try to create an one in non-voter mode.
+    #[tracing::instrument(level = "debug", skip_all)]
+    pub async fn open_create_boot(
+        config: &RaftConfig,
+        open: Option<()>,
+        create: Option<()>,
+        initialize_cluster: Option<Node>,
+    ) -> Result<Arc<MetaNode>, MetaStartupError> {
+        let mn = Self::open_create(config, open, create).await?;
 
         if let Some(node) = initialize_cluster {
             mn.init_cluster(node).await?;
@@ -718,47 +727,43 @@ impl MetaNode {
     async fn do_start(conf: &MetaConfig) -> Result<Arc<MetaNode>, MetaStartupError> {
         let raft_conf = &conf.raft_config;
 
-        let initialize_cluster = if raft_conf.single {
-            Some(conf.get_node())
-        } else {
-            None
-        };
-
         if raft_conf.single {
-            let mn = MetaNode::open_create_boot(raft_conf, Some(()), Some(()), initialize_cluster)
-                .await?;
+            let mn = MetaNode::open_create(raft_conf, Some(()), Some(())).await?;
+            mn.init_cluster(conf.get_node()).await?;
             return Ok(mn);
         }
 
         if !raft_conf.join.is_empty() {
             // Bring up a new node, join it into a cluster
 
-            let mn = MetaNode::open_create_boot(raft_conf, Some(()), Some(()), initialize_cluster)
-                .await?;
+            let mn = MetaNode::open_create(raft_conf, Some(()), Some(())).await?;
             return Ok(mn);
         }
         // open mode
 
-        let mn = MetaNode::open_create_boot(raft_conf, Some(()), None, initialize_cluster).await?;
+        let mn = MetaNode::open_create(raft_conf, Some(()), None).await?;
         Ok(mn)
     }
 
     /// Boot up the first node to create a cluster.
     /// For every cluster this func should be called exactly once.
     #[tracing::instrument(level = "debug", skip(config), fields(config_id=config.raft_config.config_id.as_str()))]
     pub async fn boot(config: &MetaConfig) -> Result<Arc<MetaNode>, MetaStartupError> {
-        let mn =
-            Self::open_create_boot(&config.raft_config, None, Some(()), Some(config.get_node()))
-                .await?;
-
+        let mn = Self::open_create(&config.raft_config, None, Some(())).await?;
+        mn.init_cluster(config.get_node()).await?;
         Ok(mn)
     }
 
-    // Initialized a single node cluster by:
-    // - Initializing raft membership.
-    // - Adding current node into the meta data.
+    /// Initialized a single node cluster if this node is just created:
+    /// - Initializing raft membership.
+    /// - Adding current node into the meta data.
     #[tracing::instrument(level = "debug", skip(self))]
     pub async fn init_cluster(&self, node: Node) -> Result<(), MetaStartupError> {
+        if self.is_opened() {
+            info!("It is opened, skip initializing cluster");
+            return Ok(());
+        }
+
         let node_id = self.sto.id;
 
         let mut cluster_node_ids = BTreeSet::new();

src/meta/service/src/watcher/watcher_manager.rs

@@ -23,8 +23,8 @@ use common_meta_types::protobuf::watch_request::FilterType;
 use common_meta_types::protobuf::Event;
 use common_meta_types::protobuf::WatchRequest;
 use common_meta_types::protobuf::WatchResponse;
+use common_meta_types::Change;
 use common_meta_types::PbSeqV;
-use common_meta_types::SeqV;
 use prost::Message;
 use tonic::Status;
 use tracing::info;
@@ -39,13 +39,6 @@ pub type WatcherStreamSender = Sender<Result<WatchResponse, Status>>;
 
 type CreateWatcherEvent = (WatchRequest, WatcherStreamSender);
 
-#[derive(Clone, Debug)]
-pub struct StateMachineKvData {
-    pub key: String,
-    pub prev: Option<SeqV>,
-    pub current: Option<SeqV>,
-}
-
 #[derive(Clone, Debug)]
 pub struct WatcherStateMachineSubscriber {
     event_tx: mpsc::UnboundedSender<WatcherEvent>,
@@ -54,7 +47,7 @@ pub struct WatcherStateMachineSubscriber {
 #[derive(Clone)]
 pub enum WatcherEvent {
     CreateWatcherEvent(CreateWatcherEvent),
-    StateMachineKvDataEvent(StateMachineKvData),
+    StateMachineKvDataEvent(Change<Vec<u8>, String>),
 }
 
 #[derive(Debug)]
@@ -106,8 +99,8 @@ impl WatcherManagerCore {
                     WatcherEvent::CreateWatcherEvent((req, tx)) => {
                         self.create_watcher_stream(req, tx).await;
                     }
-                    WatcherEvent::StateMachineKvDataEvent(kv) => {
-                        self.notify_event(kv).await;
+                    WatcherEvent::StateMachineKvDataEvent(kv_change) => {
+                        self.notify_event(kv_change).await;
                     }
                 }
             } else {
@@ -124,14 +117,15 @@ impl WatcherManagerCore {
         server_metrics::incr_watchers(-1);
     }
 
-    async fn notify_event(&mut self, kv: StateMachineKvData) {
-        let set = self.watcher_range_map.get_by_point(&kv.key);
+    async fn notify_event(&mut self, change: Change<Vec<u8>, String>) {
+        let k = change.ident.as_ref().unwrap();
+        let set = self.watcher_range_map.get_by_point(k);
         if set.is_empty() {
             return;
         }
 
-        let current = kv.current;
-        let prev = kv.prev;
+        let current = change.result;
+        let prev = change.prev;
 
         let is_delete_event = current.is_none();
         let mut remove_range_keys: Vec<RangeMapKey<String, WatcherId>> = vec![];
@@ -151,7 +145,7 @@ impl WatcherManagerCore {
             assert_eq!(stream.id, watcher_id);
             let resp = WatchResponse {
                 event: Some(Event {
-                    key: kv.key.clone(),
+                    key: k.to_string(),
                     current: current.clone().map(PbSeqV::from),
                     prev: prev.clone().map(PbSeqV::from),
                 }),
@@ -217,13 +211,9 @@ impl WatcherManagerCore {
 }
 
 impl StateMachineSubscriber for WatcherStateMachineSubscriber {
-    fn kv_changed(&self, key: &str, prev: Option<SeqV>, current: Option<SeqV>) {
+    fn kv_changed(&self, change: Change<Vec<u8>, String>) {
         let _ = self
             .event_tx
-            .send(WatcherEvent::StateMachineKvDataEvent(StateMachineKvData {
-                key: key.to_string(),
-                prev,
-                current,
-            }));
+            .send(WatcherEvent::StateMachineKvDataEvent(change));
     }
 }

src/meta/service/tests/it/grpc/metasrv_grpc_kv_api_restart_cluster.rs

@@ -238,10 +238,9 @@ async fn test_kv_api_restart_cluster_token_expired() -> anyhow::Result<()> {
 
     Ok(())
 }
+
 // Election timeout is 8~12 sec.
 // A raft node waits for a interval of election timeout before starting election
-// TODO: the raft should set the wait time by election_timeout.
-//       For now, just use a large timeout.
 fn timeout() -> Option<Duration> {
     Some(Duration::from_millis(30_000))
 }