refactor(meta-service): move kvapi-test-suite to separate crate (#18279)

* chore(meta-service): minor refine on sequence generator

* chore: udpate meta-client requirements

* refactor(meta-service): move kvapi-test-suite to separate crate

Author: drmingdrmer

Cargo.lock

@@ -98,6 +98,7 @@ members = [
     "src/meta/control",
     "src/meta/ee",
     "src/meta/kvapi",
+    "src/meta/kvapi-test-suite",
     "src/meta/process",
     "src/meta/proto-conv",
     "src/meta/protos",
@@ -149,6 +150,7 @@ databend-common-meta-cache = { path = "src/meta/cache" }
 databend-common-meta-client = { path = "src/meta/client" }
 databend-common-meta-control = { path = "src/meta/control" }
 databend-common-meta-kvapi = { path = "src/meta/kvapi" }
+databend-common-meta-kvapi-test-suite = { path = "src/meta/kvapi-test-suite" }
 databend-common-meta-process = { path = "src/meta/process" }
 databend-common-meta-raft-store = { path = "src/meta/raft-store" }
 databend-common-meta-semaphore = { path = "src/meta/semaphore" }

Cargo.toml

@@ -112,9 +112,14 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SequenceApi for KV {
         // V1 sequence stores the value in a separate key.
 
         let storage_ident = SequenceStorageIdent::new_from(name_ident.clone());
-        let x = self.get_pb(&storage_ident).await?;
+        let storage_value = self.get_pb(&storage_ident).await?;
 
-        let next_available = x.map(|seqv| *seqv.data.deref()).unwrap_or(0);
+        // If the storage value is removed, the sequence meta must also be removed.
+        let Some(storage_value) = storage_value else {
+            return Ok(None);
+        };
+
+        let next_available = *storage_value.data.deref();
 
         seq_meta.data.current = next_available;
         Ok(Some(seq_meta))
@@ -216,7 +221,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SequenceApi for KV {
                     });
                 }
                 Err(_e) => {
-                    // continue
+                    // conflict, continue
                 }
             }
         }

src/meta/api/src/sequence_api_impl.rs

@@ -57,8 +57,6 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
     ) -> Result<Result<(u64, u64), &'static str>, KVAppError> {
         debug!("{}", func_name!());
 
-        let sequence_seq = self.sequence_meta.seq;
-
         let start = self.sequence_meta.current;
         if u64::MAX - start < count {
             return Err(KVAppError::AppError(AppError::SequenceError(
@@ -77,7 +75,7 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
         // update meta
         self.sequence_meta.current += count;
 
-        let condition = vec![txn_cond_eq_seq(&self.ident, sequence_seq)];
+        let condition = vec![txn_cond_eq_seq(&self.ident, self.sequence_meta.seq)];
         let if_then = vec![
             txn_op_put_pb(&self.ident, &self.sequence_meta.data, None)?, // name -> meta
         ];
@@ -111,14 +109,14 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
         let storage_ident = SequenceStorageIdent::new_from(self.ident.clone());
         let storage_key = storage_ident.to_string_key();
 
-        let sequence_seq = self.sequence_meta.seq;
-        let sequence_meta = self.sequence_meta.data.clone();
+        let sequence_meta = &self.sequence_meta.data;
 
         let delta = count * (self.sequence_meta.step as u64);
 
-        let txn = TxnRequest::new(vec![txn_cond_eq_seq(&self.ident, sequence_seq)], vec![
-            TxnOp::fetch_add_u64(&storage_key, delta as i64),
-        ]);
+        let txn = TxnRequest::new(
+            vec![txn_cond_eq_seq(&self.ident, self.sequence_meta.seq)],
+            vec![TxnOp::fetch_add_u64(&storage_key, delta as i64)],
+        );
 
         let (succ, responses) = send_txn(self.kv_api, txn).await?;
 

src/meta/api/src/sequence_nextval_impl.rs

@@ -156,6 +156,9 @@ pub static METACLI_COMMIT_SEMVER: LazyLock<Version> = LazyLock::new(|| {
 /// - 2025-06-26: since TODO: add when merge
 ///   🖥 server: add `FetchAddU64.match_seq`
 ///
+/// - 2025-07-01: since TODO: add when enables sequence v1
+///   👥 client: new sequence API depends on `FetchAddU64`.
+///
 /// Server feature set:
 /// ```yaml
 /// server_features:

src/meta/client/src/lib.rs

@@ -0,0 +1,21 @@
+[package]
+name = "databend-common-meta-kvapi-test-suite"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = { workspace = true }
+databend-common-meta-kvapi = { workspace = true }
+databend-common-meta-types = { workspace = true }
+display-more = { workspace = true }
+fastrace = { workspace = true }
+log = { workspace = true }
+tokio = { workspace = true }
+
+[lints]
+workspace = true

src/meta/kvapi-test-suite/Cargo.toml

@@ -0,0 +1,18 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![allow(clippy::uninlined_format_args)]
+
+mod test_suite;
+pub use test_suite::TestSuite;

src/meta/kvapi-test-suite/src/lib.rs

@@ -14,6 +14,7 @@
 
 use std::time::Duration;
 
+use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_types::protobuf as pb;
 use databend_common_meta_types::protobuf::BooleanExpression;
 use databend_common_meta_types::protobuf::FetchAddU64Response;
@@ -48,11 +49,9 @@ use fastrace::func_path;
 use log::debug;
 use log::info;
 
-use crate::kvapi;
-
 pub struct TestSuite {}
 
-impl kvapi::TestSuite {
+impl TestSuite {
     #[fastrace::trace]
     pub async fn test_all<KV, B>(&self, builder: B) -> anyhow::Result<()>
     where
@@ -115,7 +114,7 @@ impl kvapi::TestSuite {
     }
 }
 
-impl kvapi::TestSuite {
+impl TestSuite {
     #[fastrace::trace]
     pub async fn kv_write_read<KV: kvapi::KVApi>(&self, kv: &KV) -> anyhow::Result<()> {
         info!("--- kvapi::KVApiTestSuite::kv_write_read() start");
@@ -1601,7 +1600,7 @@ impl kvapi::TestSuite {
 }
 
 /// Test that write and read should be forwarded to leader
-impl kvapi::TestSuite {
+impl TestSuite {
     #[fastrace::trace]
     pub async fn kv_write_read_across_nodes<KV: kvapi::KVApi>(
         &self,

src/meta/kvapi/src/kvapi/test_suite.rs renamed to src/meta/kvapi-test-suite/src/test_suite.rs

@@ -13,13 +13,10 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 databend-common-meta-app-types = { workspace = true }
 databend-common-meta-types = { workspace = true }
-display-more = { workspace = true }
-fastrace = { workspace = true }
 futures-util = { workspace = true }
 log = { workspace = true }
 serde = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true }
 
 [lints]
 workspace = true

src/meta/kvapi/Cargo.toml

@@ -23,7 +23,6 @@ mod key_parser;
 mod message;
 mod pair;
 mod prefix;
-mod test_suite;
 mod value;
 mod value_with_name;
 
@@ -52,6 +51,5 @@ pub use pair::BasicPair;
 pub use pair::Pair;
 pub use pair::SeqPair;
 pub use prefix::prefix_to_range;
-pub use test_suite::TestSuite;
 pub use value::Value;
 pub use value_with_name::ValueWithName;