Feature: transaction api

common/management/tests/it/user.rs

@@ -29,6 +29,8 @@ use common_meta_types::Operation;
 use common_meta_types::PasswordHashMethod;
 use common_meta_types::PrefixListReply;
 use common_meta_types::SeqV;
+use common_meta_types::TransactionReply;
+use common_meta_types::TransactionReq;
 use common_meta_types::UpsertKVAction;
 use common_meta_types::UpsertKVActionReply;
 use common_meta_types::UserIdentity;
@@ -53,6 +55,9 @@ mock! {
         ) -> Result<MGetKVActionReply,MetaError>;
 
         async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError>;
+
+        async fn transaction(&self, txn: TransactionReq) -> Result<TransactionReply, MetaError>;
+
         }
 }
 

common/meta/api/src/kv_api.rs

@@ -20,6 +20,8 @@ use common_meta_types::GetKVActionReply;
 use common_meta_types::MGetKVActionReply;
 use common_meta_types::MetaError;
 use common_meta_types::PrefixListReply;
+use common_meta_types::TransactionReply;
+use common_meta_types::TransactionReq;
 use common_meta_types::UpsertKVAction;
 use common_meta_types::UpsertKVActionReply;
 
@@ -44,6 +46,8 @@ pub trait KVApi: Send + Sync {
     async fn mget_kv(&self, key: &[String]) -> Result<MGetKVActionReply, MetaError>;
 
     async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError>;
+
+    async fn transaction(&self, txn: TransactionReq) -> Result<TransactionReply, MetaError>;
 }
 
 #[async_trait]
@@ -63,4 +67,8 @@ impl<U: KVApi, T: Deref<Target = U> + Send + Sync> KVApi for T {
     async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError> {
         self.deref().prefix_list_kv(prefix).await
     }
+
+    async fn transaction(&self, txn: TransactionReq) -> Result<TransactionReply, MetaError> {
+        self.deref().transaction(txn).await
+    }
 }

common/meta/api/src/kv_api_test_suite.rs

@@ -20,6 +20,18 @@ use common_meta_types::KVMeta;
 use common_meta_types::MatchSeq;
 use common_meta_types::Operation;
 use common_meta_types::SeqV;
+use common_meta_types::TransactionCondition;
+use common_meta_types::TransactionConditionResult;
+use common_meta_types::TransactionConditionTarget;
+use common_meta_types::TransactionDeleteRequest;
+use common_meta_types::TransactionDeleteResponse;
+use common_meta_types::TransactionGetRequest;
+use common_meta_types::TransactionGetResponse;
+use common_meta_types::TransactionOperation;
+use common_meta_types::TransactionOperationResponse;
+use common_meta_types::TransactionPutRequest;
+use common_meta_types::TransactionPutResponse;
+use common_meta_types::TransactionReq;
 use common_meta_types::UpsertKVAction;
 use common_tracing::tracing;
 
@@ -42,9 +54,9 @@ impl KVApiTestSuite {
         self.kv_meta(&builder.build().await).await?;
         self.kv_list(&builder.build().await).await?;
         self.kv_mget(&builder.build().await).await?;
+        self.kv_transaction(&builder.build().await).await?;
 
         // Run cross node test on every 2 adjacent nodes
-
         let mut i = 0;
         loop {
             let cluster = builder.build_cluster().await;
@@ -56,7 +68,6 @@ impl KVApiTestSuite {
             }
             i += 1;
         }
-
         Ok(())
     }
 }
@@ -532,6 +543,244 @@ impl KVApiTestSuite {
 
         Ok(())
     }
+
+    fn check_transaction_responses(
+        &self,
+        responses: &Vec<TransactionOperationResponse>,
+        expected: &Vec<TransactionOperationResponse>,
+    ) {
+        assert_eq!(responses.len(), expected.len());
+
+        for i in 0..responses.len() {
+            let resp = &responses[i];
+            let expect_resp = &expected[i];
+
+            assert_eq!(resp, expect_resp);
+        }
+    }
+
+    //#[tracing::instrument(level = "info", skip(self, kv))]
+    pub async fn kv_transaction<KV: KVApi>(&self, kv: &KV) -> anyhow::Result<()> {
+        // first case: get and set one key transaction
+        {
+            let k1 = "txn_1_K1";
+            let val1 = b"v1".to_vec();
+
+            // first insert k1 value
+            kv.upsert_kv(UpsertKVAction::new(
+                k1,
+                MatchSeq::Any,
+                Operation::Update(val1.clone()),
+                None,
+            ))
+            .await?;
+
+            // transaction by k1 condition
+            let txn_key = k1.to_string();
+            let condition = vec![TransactionCondition {
+                key: txn_key.clone(),
+                target: TransactionConditionTarget::Key,
+                expected: TransactionConditionResult::KeyExist,
+                value: None,
+            }];
+
+            let if_then: Vec<TransactionOperation> =
+                vec![TransactionOperation::TransactionPutRequest(
+                    TransactionPutRequest {
+                        key: txn_key.clone(),
+                        value: b"new_v1".to_vec(),
+                        prev_kv: true,
+                    },
+                )];
+
+            let else_then: Vec<TransactionOperation> = vec![];
+            let txn = TransactionReq {
+                condition,
+                if_then,
+                else_then,
+            };
+
+            let resp = kv.transaction(txn).await?;
+
+            assert!(resp.success);
+            let expected: Vec<TransactionOperationResponse> =
+                vec![TransactionOperationResponse::TransactionPutResponse(
+                    TransactionPutResponse {
+                        key: txn_key.clone(),
+                        prev_value: Some(SeqV::with_meta(1, None, val1.clone())),
+                    },
+                )];
+
+            self.check_transaction_responses(&resp.responses, &expected);
+        }
+
+        // second case: get two key(one not exist) and set one key transaction
+        {
+            // first insert k1 value
+            let k1 = "txn_2_K1";
+            let k2 = "txn_2_K2";
+
+            kv.upsert_kv(UpsertKVAction::new(
+                k1,
+                MatchSeq::Any,
+                Operation::Update(b"v1".to_vec()),
+                None,
+            ))
+            .await?;
+
+            // transaction by k1 and k2 condition
+            let txn_key1 = k1.to_string();
+            let txn_key2 = k2.to_string();
+            let condition = vec![
+                TransactionCondition {
+                    key: txn_key1.clone(),
+                    target: TransactionConditionTarget::Key,
+                    expected: TransactionConditionResult::KeyExist,
+                    value: None,
+                },
+                TransactionCondition {
+                    key: txn_key2.clone(),
+                    target: TransactionConditionTarget::Key,
+                    expected: TransactionConditionResult::KeyExist,
+                    value: None,
+                },
+            ];
+
+            let if_then: Vec<TransactionOperation> =
+                vec![TransactionOperation::TransactionPutRequest(
+                    TransactionPutRequest {
+                        key: txn_key1.clone(),
+                        value: b"new_v1".to_vec(),
+                        prev_kv: true,
+                    },
+                )];
+
+            let else_then: Vec<TransactionOperation> = vec![];
+            let txn = TransactionReq {
+                condition,
+                if_then,
+                else_then,
+            };
+
+            let resp = kv.transaction(txn).await?;
+
+            // k2 not exists, so the resp MUST return false
+            assert!(!resp.success);
+            assert_eq!(resp.responses.len(), 0);
+        }
+
+        // 3rd case: get two key and set both key transaction
+        {
+            let k1 = "txn_3_K1";
+            let val1 = b"v1".to_vec();
+            let val1_new = b"v1_new".to_vec();
+
+            let k2 = "txn_3_K2";
+            let val2 = b"v1".to_vec();
+
+            // first insert k1 and k2 value
+            kv.upsert_kv(UpsertKVAction::new(
+                k1,
+                MatchSeq::Any,
+                Operation::Update(val1.clone()),
+                None,
+            ))
+            .await?;
+            kv.upsert_kv(UpsertKVAction::new(
+                k2,
+                MatchSeq::Any,
+                Operation::Update(val2.clone()),
+                None,
+            ))
+            .await?;
+
+            // transaction by k1 and k2 condition
+            let txn_key1 = k1.to_string();
+            let txn_key2 = k2.to_string();
+
+            let condition = vec![
+                TransactionCondition {
+                    key: txn_key1.clone(),
+                    target: TransactionConditionTarget::Key,
+                    expected: TransactionConditionResult::KeyExist,
+                    value: None,
+                },
+                TransactionCondition {
+                    key: txn_key2.clone(),
+                    target: TransactionConditionTarget::Key,
+                    expected: TransactionConditionResult::KeyExist,
+                    value: None,
+                },
+            ];
+
+            let if_then: Vec<TransactionOperation> = vec![
+                // change k1
+                TransactionOperation::TransactionPutRequest(TransactionPutRequest {
+                    key: txn_key1.clone(),
+                    value: val1_new.clone(),
+                    prev_kv: true,
+                }),
+                // change k2
+                TransactionOperation::TransactionPutRequest(TransactionPutRequest {
+                    key: txn_key2.clone(),
+                    value: b"new_v2".to_vec(),
+                    prev_kv: true,
+                }),
+                // get k1
+                TransactionOperation::TransactionGetRequest(TransactionGetRequest {
+                    key: txn_key1.clone(),
+                }),
+                // delete k1
+                TransactionOperation::TransactionDeleteRequest(TransactionDeleteRequest {
+                    key: txn_key1.clone(),
+                    prev_kv: true,
+                }),
+                // get k1
+                TransactionOperation::TransactionGetRequest(TransactionGetRequest {
+                    key: txn_key1.clone(),
+                }),
+            ];
+
+            let else_then: Vec<TransactionOperation> = vec![];
+            let txn = TransactionReq {
+                condition,
+                if_then,
+                else_then,
+            };
+
+            let resp = kv.transaction(txn).await?;
+
+            assert!(resp.success);
+            let expected: Vec<TransactionOperationResponse> = vec![
+                TransactionOperationResponse::TransactionPutResponse(TransactionPutResponse {
+                    key: txn_key1.clone(),
+                    prev_value: Some(SeqV::with_meta(4, None, val1.clone())),
+                }),
+                TransactionOperationResponse::TransactionPutResponse(TransactionPutResponse {
+                    key: txn_key2.clone(),
+                    prev_value: Some(SeqV::with_meta(5, None, val2.clone())),
+                }),
+                TransactionOperationResponse::TransactionGetResponse(TransactionGetResponse {
+                    key: txn_key1.clone(),
+                    value: Some(SeqV::with_meta(6, None, val1_new.clone())),
+                }),
+                TransactionOperationResponse::TransactionDeleteResponse(
+                    TransactionDeleteResponse {
+                        key: txn_key1.clone(),
+                        success: true,
+                        prev_value: Some(SeqV::with_meta(6, None, val1_new.clone())),
+                    },
+                ),
+                TransactionOperationResponse::TransactionGetResponse(TransactionGetResponse {
+                    key: txn_key1.clone(),
+                    value: None,
+                }),
+            ];
+
+            self.check_transaction_responses(&resp.responses, &expected);
+        }
+        Ok(())
+    }
 }
 
 /// Test that write and read should be forwarded to leader

common/meta/embedded/src/kv_api_impl.rs

@@ -19,6 +19,8 @@ use common_meta_types::GetKVActionReply;
 use common_meta_types::MGetKVActionReply;
 use common_meta_types::MetaError;
 use common_meta_types::PrefixListReply;
+use common_meta_types::TransactionReply;
+use common_meta_types::TransactionReq;
 use common_meta_types::UpsertKVAction;
 use common_meta_types::UpsertKVActionReply;
 
@@ -45,4 +47,9 @@ impl KVApi for MetaEmbedded {
         let sm = self.inner.lock().await;
         sm.prefix_list_kv(prefix).await
     }
+
+    async fn transaction(&self, txn: TransactionReq) -> Result<TransactionReply, MetaError> {
+        let sm = self.inner.lock().await;
+        sm.transaction(txn).await
+    }
 }