feat(meta-service): return current seq-value in txn-put response (#18136)

* feat(meta-service): return current seq-value in txn-put response

TxnPutResponse now returns both the value before and after the put
operation. Each returned value is a `SeqV` struct containing the
sequence number and value in bytes.

This is a backward-compatible change as protobuf-based clients
will ignore the new field.

* chore: normalize values before compare for kv-api tests

Author: drmingdrmer

src/meta/client/src/lib.rs

@@ -144,9 +144,13 @@ pub static METACLI_COMMIT_SEMVER: LazyLock<Version> = LazyLock::new(|| {
 /// - 2025-05-08: since 1.2.736
 ///   🖥 server: add `WatchResponse::is_initialization`,
 ///
-/// - 2025-06-09: since TODO: update when merged
+/// - 2025-06-09: since 1.2.755
 ///   🖥 server: remove `TxnReply::error`
 ///
+/// - 2025-06-11: since TODO: update when merge
+///   🖥 server: add `TxnPutResponse::current`
+///
+///
 /// Server feature set:
 /// ```yaml
 /// server_features:

src/meta/kvapi/src/kvapi/test_suite.rs

@@ -491,8 +491,12 @@ impl kvapi::TestSuite {
         let responses = &reply.responses;
         assert_eq!(responses.len(), expected.len());
 
+        let responses = normalize_txn_response(responses.clone());
+        let expected = normalize_txn_response(expected.to_vec());
+
         for i in 0..responses.len() {
             let resp = &responses[i];
+
             let expect_resp = &expected[i];
 
             assert_eq!(resp, expect_resp, "{}-th response", i);
@@ -521,6 +525,7 @@ impl kvapi::TestSuite {
             response: Some(txn_op_response::Response::Put(TxnPutResponse {
                 key: txn_key.clone(),
                 prev_value: None,
+                current: Some(pb::SeqV::new(1, b"new_v1".to_vec())),
             })),
         }];
 
@@ -659,6 +664,7 @@ impl kvapi::TestSuite {
                 response: Some(txn_op_response::Response::Put(TxnPutResponse {
                     key: txn_key.clone(),
                     prev_value: Some(pb::SeqV::from(SeqV::new(1, val1.clone()))),
+                    current: Some(pb::SeqV::new(2, b("new_v1"))),
                 })),
             }];
 
@@ -766,14 +772,16 @@ impl kvapi::TestSuite {
                 TxnOpResponse {
                     response: Some(txn_op_response::Response::Put(TxnPutResponse {
                         key: txn_key1.clone(),
-                        prev_value: Some(pb::SeqV::from(SeqV::new(4, val1.clone()))),
+                        prev_value: Some(pb::SeqV::new(4, val1.clone())),
+                        current: Some(pb::SeqV::new(6, val1_new.clone())),
                     })),
                 },
                 // change k2
                 TxnOpResponse {
                     response: Some(txn_op_response::Response::Put(TxnPutResponse {
                         key: txn_key2.clone(),
-                        prev_value: Some(pb::SeqV::from(SeqV::new(5, val2.clone()))),
+                        prev_value: Some(pb::SeqV::new(5, val2.clone())),
+                        current: Some(pb::SeqV::new(7, b("new_v2").clone())),
                     })),
                 },
                 // get k1
@@ -813,35 +821,6 @@ impl kvapi::TestSuite {
 
         // 4th case: get one key by value and set key transaction
         {
-            /// Convert Some(KvMeta{ expire: None }) to None to simplify the comparison
-            fn norm(vs: Vec<TxnOpResponse>) -> Vec<TxnOpResponse> {
-                vs.into_iter()
-                    .map(|mut v| {
-                        //
-                        match &mut v.response {
-                            Some(Response::Get(TxnGetResponse {
-                                value: Some(pb::SeqV { meta, .. }),
-                                ..
-                            })) => {
-                                if *meta == Some(pb::KvMeta { expire_at: None }) {
-                                    *meta = None;
-                                }
-                            }
-                            Some(Response::Put(TxnPutResponse {
-                                prev_value: Some(pb::SeqV { meta, .. }),
-                                ..
-                            })) => {
-                                if *meta == Some(pb::KvMeta { expire_at: None }) {
-                                    *meta = None;
-                                }
-                            }
-                            _ => {}
-                        }
-                        v
-                    })
-                    .collect()
-            }
-
             let k1 = "txn_4_K1";
 
             kv.upsert_kv(UpsertKV::update(k1, b"v1")).await?;
@@ -872,12 +851,19 @@ impl kvapi::TestSuite {
             let resp = kv.transaction(txn).await?;
 
             let expected: Vec<TxnOpResponse> = vec![
-                TxnOpResponse::put(k1, Some(pb::SeqV::new(8, b("v1")))),
+                TxnOpResponse::put(
+                    k1,
+                    Some(pb::SeqV::new(8, b("v1"))),
+                    Some(pb::SeqV::new(9, b("v2"))),
+                ),
                 TxnOpResponse::get(k1, Some(SeqV::new(9, b("v2")))),
             ];
 
             assert_eq!(resp.success, true);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test less than value: success = false
 
@@ -893,7 +879,10 @@ impl kvapi::TestSuite {
                 vec![TxnOpResponse::get(k1, Some(SeqV::new(9, b("v2"))))];
 
             assert_eq!(resp.success, false);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test less than value: success = true
 
@@ -906,12 +895,19 @@ impl kvapi::TestSuite {
             let resp = kv.transaction(txn).await?;
 
             let expected: Vec<TxnOpResponse> = vec![
-                TxnOpResponse::put(k1, Some(pb::SeqV::new(9, b("v2")))),
+                TxnOpResponse::put(
+                    k1,
+                    Some(pb::SeqV::new(9, b("v2"))),
+                    Some(pb::SeqV::new(10, b("v3"))),
+                ),
                 TxnOpResponse::get(k1, Some(SeqV::new(10, b("v3")))),
             ];
 
             assert_eq!(resp.success, true);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test less equal value: success = false
 
@@ -927,7 +923,10 @@ impl kvapi::TestSuite {
                 vec![TxnOpResponse::get(k1, Some(SeqV::new(10, b("v3"))))];
 
             assert_eq!(resp.success, false);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test less equal value: success = true
 
@@ -940,12 +939,19 @@ impl kvapi::TestSuite {
             let resp = kv.transaction(txn).await?;
 
             let expected: Vec<TxnOpResponse> = vec![
-                TxnOpResponse::put(k1, Some(pb::SeqV::new(10, b("v3")))),
+                TxnOpResponse::put(
+                    k1,
+                    Some(pb::SeqV::new(10, b("v3"))),
+                    Some(pb::SeqV::new(11, b("v4"))),
+                ),
                 TxnOpResponse::get(k1, Some(SeqV::new(11, b("v4")))),
             ];
 
             assert_eq!(resp.success, true);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test greater than value: success = false
 
@@ -961,7 +967,10 @@ impl kvapi::TestSuite {
                 vec![TxnOpResponse::get(k1, Some(SeqV::new(11, b("v4"))))];
 
             assert_eq!(resp.success, false);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test greater than value: success = true
 
@@ -974,12 +983,19 @@ impl kvapi::TestSuite {
             let resp = kv.transaction(txn).await?;
 
             let expected: Vec<TxnOpResponse> = vec![
-                TxnOpResponse::put(k1, Some(pb::SeqV::new(11, b("v4")))),
+                TxnOpResponse::put(
+                    k1,
+                    Some(pb::SeqV::new(11, b("v4"))),
+                    Some(pb::SeqV::new(12, b("v5"))),
+                ),
                 TxnOpResponse::get(k1, Some(SeqV::new(12, b("v5")))),
             ];
 
             assert_eq!(resp.success, true);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test greater equal value: success = false
 
@@ -995,7 +1011,10 @@ impl kvapi::TestSuite {
                 vec![TxnOpResponse::get(k1, Some(SeqV::new(12, b("v5"))))];
 
             assert_eq!(resp.success, false);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
 
             // Test greater equal value: success = true
 
@@ -1008,12 +1027,19 @@ impl kvapi::TestSuite {
             let resp = kv.transaction(txn).await?;
 
             let expected: Vec<TxnOpResponse> = vec![
-                TxnOpResponse::put(k1, Some(pb::SeqV::new(12, b("v5")))),
+                TxnOpResponse::put(
+                    k1,
+                    Some(pb::SeqV::new(12, b("v5"))),
+                    Some(pb::SeqV::new(13, b("v6"))),
+                ),
                 TxnOpResponse::get(k1, Some(SeqV::new(13, b("v6")))),
             ];
 
             assert_eq!(resp.success, true);
-            assert_eq!(norm(resp.responses), norm(expected));
+            assert_eq!(
+                normalize_txn_response(resp.responses),
+                normalize_txn_response(expected)
+            );
         }
         Ok(())
     }
@@ -1410,6 +1436,48 @@ impl kvapi::TestSuite {
     }
 }
 
+/// Convert Some(KvMeta{ expire: None }) to None to simplify the comparison
+fn normalize_txn_response(vs: Vec<TxnOpResponse>) -> Vec<TxnOpResponse> {
+    vs.into_iter()
+        .map(|mut v| {
+            //
+            match &mut v.response {
+                Some(Response::Get(TxnGetResponse {
+                    value: Some(pb::SeqV { meta, .. }),
+                    ..
+                })) => {
+                    if *meta == Some(pb::KvMeta { expire_at: None }) {
+                        *meta = None;
+                    }
+                }
+                Some(Response::Put(TxnPutResponse {
+                    prev_value: Some(pb::SeqV { meta, .. }),
+                    ..
+                })) => {
+                    if *meta == Some(pb::KvMeta { expire_at: None }) {
+                        *meta = None;
+                    }
+                }
+                _ => {}
+            }
+
+            match &mut v.response {
+                Some(Response::Put(TxnPutResponse {
+                    current: Some(pb::SeqV { meta, .. }),
+                    ..
+                })) => {
+                    if *meta == Some(pb::KvMeta { expire_at: None }) {
+                        *meta = None;
+                    }
+                }
+                _ => {}
+            }
+
+            v
+        })
+        .collect()
+}
+
 fn b(x: impl ToString) -> Vec<u8> {
     x.to_string().as_bytes().to_vec()
 }

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

@@ -470,11 +470,12 @@ where SM: StateMachineApi + 'static
             put.ttl_ms.map(Interval::from_millis),
         ));
 
-        let (prev, _result) = self.upsert_kv(&upsert).await?;
+        let (prev, result) = self.upsert_kv(&upsert).await?;
 
         let put_resp = TxnPutResponse {
             key: put.key.clone(),
             prev_value: prev.map(pb::SeqV::from),
+            current: result.map(pb::SeqV::from),
         };
 
         resp.responses.push(TxnOpResponse {

src/meta/types/proto/request.proto

@@ -59,7 +59,12 @@ message TxnPutRequest {
 
 message TxnPutResponse {
   string key = 1;
+
+  // The value before put
   optional SeqV prev_value = 2;
+
+  // The value after put
+  optional SeqV current = 3;
 }
 
 // Delete request and response
@@ -124,7 +129,7 @@ message TxnDeleteResponse {
 }
 
 // Delete by prefix request and response
-message TxnDeleteByPrefixRequest { string prefix = 1; }
+message TxnDeleteByPrefixRequest {string prefix = 1;}
 
 message TxnDeleteByPrefixResponse {
   string prefix = 1;

src/meta/types/src/proto_display/mod.rs

@@ -270,9 +270,10 @@ impl Display for TxnPutResponse {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(
             f,
-            "Put-resp: key={}, prev_seq={:?}",
+            "Put-resp: key={}, prev_seq={}, current_seq={}",
             self.key,
-            self.prev_value.as_ref().map(|x| x.seq)
+            self.prev_value.as_ref().map(|x| x.seq).display(),
+            self.current.as_ref().map(|x| x.seq).display()
         )
     }
 }

src/meta/types/src/proto_ext/txn_ext.rs

@@ -329,11 +329,16 @@ impl pb::TxnOpResponse {
     }
 
     /// Create a new `TxnOpResponse` of a `Put` operation.
-    pub fn put(key: impl ToString, prev_value: Option<pb::SeqV>) -> Self {
+    pub fn put(
+        key: impl ToString,
+        prev_value: Option<pb::SeqV>,
+        current: Option<pb::SeqV>,
+    ) -> Self {
         pb::TxnOpResponse {
             response: Some(pb::txn_op_response::Response::Put(pb::TxnPutResponse {
                 key: key.to_string(),
                 prev_value,
+                current,
             })),
         }
     }