refactor: start a meta-service as local meta for testing

- Replaced the in-memory meta-store with a meta-service running in a
  temporary directory for testing purposes.

- The in-memory meta-store was limited in functionality and did not
  provide a complete feature set.

- The new approach ensures full functionality during testing by
  simulating a real meta-service.

Author: drmingdrmer

Cargo.lock

@@ -11,16 +11,20 @@ edition = { workspace = true }
 io-uring = []
 
 [dependencies]
+anyhow = { workspace = true }
 async-trait = { workspace = true }
 databend-common-base = { workspace = true }
 databend-common-grpc = { workspace = true }
 databend-common-meta-client = { workspace = true }
-databend-common-meta-embedded = { workspace = true }
 databend-common-meta-kvapi = { workspace = true }
 databend-common-meta-semaphore = { workspace = true }
 databend-common-meta-types = { workspace = true }
+databend-meta = { workspace = true }
 log = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true }
 tokio-stream = { workspace = true }
+tonic = { workspace = true }
 
 [lints]
 workspace = true

src/meta/store/Cargo.toml

@@ -12,33 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::hash_map::Entry;
+pub(crate) mod local;
+
+use std::ops::Deref;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
 use std::time::Duration;
 
-use databend_common_base::base::tokio::sync::Semaphore as TokioSemaphore;
 use databend_common_grpc::RpcClientConf;
 use databend_common_meta_client::errors::CreationError;
 use databend_common_meta_client::ClientHandle;
 use databend_common_meta_client::MetaGrpcClient;
-use databend_common_meta_embedded::MemMeta;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::KVStream;
 use databend_common_meta_kvapi::kvapi::UpsertKVReply;
 use databend_common_meta_semaphore::acquirer::Permit;
-use databend_common_meta_semaphore::acquirer::SharedAcquirerStat;
 use databend_common_meta_semaphore::errors::AcquireError;
-use databend_common_meta_semaphore::errors::ConnectionClosed;
 use databend_common_meta_semaphore::Semaphore;
 use databend_common_meta_types::protobuf::WatchRequest;
 use databend_common_meta_types::protobuf::WatchResponse;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::TxnReply;
 use databend_common_meta_types::TxnRequest;
 use databend_common_meta_types::UpsertKV;
+pub use local::LocalMetaService;
 use log::info;
 use tokio_stream::Stream;
 
@@ -53,11 +52,22 @@ pub struct MetaStoreProvider {
 /// MetaStore is impl with either a local embedded meta store, or a grpc-client of metasrv
 #[derive(Clone)]
 pub enum MetaStore {
-    L(Arc<MemMeta>),
+    L(Arc<LocalMetaService>),
     R(Arc<ClientHandle>),
 }
 
 impl MetaStore {
+    /// Create a local meta service for testing.
+    ///
+    /// It is required to assign a base port as the port number range.
+    pub async fn new_local_testing(base_port: u16) -> Self {
+        MetaStore::L(Arc::new(
+            LocalMetaService::new("MetaStore-new-local-testing", Some(base_port))
+                .await
+                .unwrap(),
+        ))
+    }
+
     pub fn arc(self) -> Arc<Self> {
         Arc::new(self)
     }
@@ -70,23 +80,23 @@ impl MetaStore {
     }
 
     pub async fn get_local_addr(&self) -> std::result::Result<Option<String>, MetaError> {
-        match self {
-            MetaStore::L(_) => Ok(None),
-            MetaStore::R(grpc_client) => {
-                let client_info = grpc_client.get_client_info().await?;
-                Ok(Some(client_info.client_addr))
-            }
-        }
+        let client = match self {
+            MetaStore::L(l) => l.deref().deref(),
+            MetaStore::R(grpc_client) => grpc_client,
+        };
+
+        let client_info = client.get_client_info().await?;
+        Ok(Some(client_info.client_addr))
     }
 
     pub async fn watch(&self, request: WatchRequest) -> Result<WatchStream, MetaError> {
-        match self {
-            MetaStore::L(_) => unreachable!(),
-            MetaStore::R(grpc_client) => {
-                let streaming = grpc_client.request(request).await?;
-                Ok(Box::pin(WatchResponseStream::create(streaming)))
-            }
-        }
+        let client = match self {
+            MetaStore::L(l) => l.deref(),
+            MetaStore::R(grpc_client) => grpc_client,
+        };
+
+        let streaming = client.request(request).await?;
+        Ok(Box::pin(WatchResponseStream::create(streaming)))
     }
 
     pub async fn new_acquired(
@@ -96,34 +106,12 @@ impl MetaStore {
         id: impl ToString,
         lease: Duration,
     ) -> Result<Permit, AcquireError> {
-        match self {
-            MetaStore::L(v) => {
-                let mut local_lock_map = v.locks.lock().await;
-
-                let acquire_res = match local_lock_map.entry(prefix.to_string()) {
-                    Entry::Occupied(v) => v.get().clone(),
-                    Entry::Vacant(v) => v
-                        .insert(Arc::new(TokioSemaphore::new(capacity as usize)))
-                        .clone(),
-                };
+        let client = match self {
+            MetaStore::L(l) => l.deref(),
+            MetaStore::R(grpc_client) => grpc_client,
+        };
 
-                match acquire_res.acquire_owned().await {
-                    Ok(guard) => Ok(Permit {
-                        stat: SharedAcquirerStat::new(),
-                        fu: Box::pin(async move {
-                            let _guard = guard;
-                            Ok(())
-                        }),
-                    }),
-                    Err(_e) => Err(AcquireError::ConnectionClosed(ConnectionClosed::new_str(
-                        "",
-                    ))),
-                }
-            }
-            MetaStore::R(grpc_client) => {
-                Semaphore::new_acquired(grpc_client.clone(), prefix, capacity, id, lease).await
-            }
-        }
+        Semaphore::new_acquired(client.clone(), prefix, capacity, id, lease).await
     }
 
     pub async fn new_acquired_by_time(
@@ -133,35 +121,12 @@ impl MetaStore {
         id: impl ToString,
         lease: Duration,
     ) -> Result<Permit, AcquireError> {
-        match self {
-            MetaStore::L(v) => {
-                let mut local_lock_map = v.locks.lock().await;
+        let client = match self {
+            MetaStore::L(l) => l.deref(),
+            MetaStore::R(grpc_client) => grpc_client,
+        };
 
-                let acquire_res = match local_lock_map.entry(prefix.to_string()) {
-                    Entry::Occupied(v) => v.get().clone(),
-                    Entry::Vacant(v) => v
-                        .insert(Arc::new(TokioSemaphore::new(capacity as usize)))
-                        .clone(),
-                };
-
-                match acquire_res.acquire_owned().await {
-                    Ok(guard) => Ok(Permit {
-                        stat: SharedAcquirerStat::new(),
-                        fu: Box::pin(async move {
-                            let _guard = guard;
-                            Ok(())
-                        }),
-                    }),
-                    Err(_e) => Err(AcquireError::ConnectionClosed(ConnectionClosed::new_str(
-                        "",
-                    ))),
-                }
-            }
-            MetaStore::R(grpc_client) => {
-                Semaphore::new_acquired_by_time(grpc_client.clone(), prefix, capacity, id, lease)
-                    .await
-            }
-        }
+        Semaphore::new_acquired_by_time(client.clone(), prefix, capacity, id, lease).await
     }
 }
 
@@ -211,8 +176,11 @@ impl MetaStoreProvider {
             );
 
             // NOTE: This can only be used for test: data will be removed when program quit.
-            let meta_store = MemMeta::default();
-            Ok(MetaStore::L(Arc::new(meta_store)))
+            Ok(MetaStore::L(Arc::new(
+                LocalMetaService::new("MetaStoreProvider-created", None)
+                    .await
+                    .unwrap(),
+            )))
         } else {
             info!(conf :? =(&self.rpc_conf); "use remote meta");
             let client = MetaGrpcClient::try_new(&self.rpc_conf)?;