Repository: databendlabs/databend (language: rust)

Issue:
bug: double lock ownership due to concurrent race
### Search before asking

- [X] I had searched in the [issues](https://github.com/databendlabs/databend/issues) and found no similar issues.


### Version

v1.2.662-nightly

### What's Wrong?

A test case reports a failure where two queries simultaneously hold locks for the same table. The root cause is that the create_lock_revision process lacks atomicity between the two operations: generating a new revision and inserting that revision into the lock key list. This non-atomic behavior allows two queries to independently acquire revisions and insert them into the list without proper synchronization, leading to both queries believing they own the lock.

### How to Reproduce?

https://github.com/databendlabs/databend/actions/runs/11947873485/job/33305193605
```
03_0016_update_with_lock:                                               [ FAIL ] - result differs with:
--- /runner/_work/databend/databend/tests/suites/0_stateless/03_dml/03_0016_update_with_lock.result	2024-11-21 06:52:54.029128224 +0000
+++ /runner/_work/databend/databend/tests/suites/0_stateless/03_dml/03_0016_update_with_lock.stdout	2024-11-21 06:54:48.250200892 +0000
@@ -1,3 +1,4 @@
 10
 Test table lock for update
-10
+Error: APIError: ResponseError with 4001: conflict resolve context:ModifiedSegmentExistsInLatest(SnapshotChanges { appended_segments: [], replaced_segments: {0: ("129/135/_sg/a6c06dcd069948bf85262397021ad89c_v4.mpk", 4)}, removed_segment_indexes: [], merged_statistics: Statistics { row_count: 10, block_count: 1, perfect_block_count: 0, uncompressed_byte_size: 84, compressed_byte_size: 590, index_size: 669, col_stats: {0: ColumnStatistics { min: Number(1_i32), max: Number(10_i32), null_count: 0, in_memory_size: 42, distinct_of_values: None }, 1: ColumnStatistics { min: Number(2_i32), max: Number(11_i32), null_count: 0, in_memory_size: 42, distinct_of_values: None }}, cluster_stats: None }, removed_statistics: Statistics { row_count: 10, block_count: 1, perfect_block_count: 0, uncompressed_byte_size: 84, compressed_byte_size: 590, index_size: 672, col_stats: {0: ColumnStatistics { min: Number(1_i32), max: Number(10_i32), null_count: 0, in_memory_size: 42, distinct_of_values: None }, 1: ColumnStatistics { min: Number(2_i32), max: Number(11_i32), null_count: 0, in_memory_size: 42, distinct_of_values: None }}, cluster_stats: None } })
+9
```
```
edf2cfb4-970b-4f1a-a27c-00cb83f11b9c 2024-11-21T06:54:47.747564Z DEBUG databend_query::locks::lock_holder: [lock_holder.rs:181](http://lock_holder.rs:181/) create table lock success, revision=157
edf2cfb4-970b-4f1a-a27c-00cb83f11b9c 2024-11-21T06:54:47.747580Z DEBUG databend_common_meta_client::grpc_client: [grpc_client.rs:353](http://grpc_client.rs:353/) ClientHandle([0.0.0.0:9191](http://0.0.0.0:9191/)) send request to meta client worker: request: ClientWorkerRequest { request_id: 1054, req: StreamList(Streamed(ListKVReq { prefix: "__fd_table_lock/test_tenant/135/" })) }
edf2cfb4-970b-4f1a-a27c-00cb83f11b9c 2024-11-21T06:54:47.747845Z DEBUG databend_common_meta_client::grpc_client: [grpc_client.rs:1116](http://grpc_client.rs:1116/) MetaGrpcClient([0.0.0.0:9191](http://0.0.0.0:9191/))::kv_read_v1 request: ListKV(ListKVReq { prefix: "__fd_table_lock/test_tenant/135/" })

b213d938-d09d-4535-8b49-5426f5cc22c9 2024-11-21T06:54:47.746641Z DEBUG databend_query::locks::lock_holder: [lock_holder.rs:181](http://lock_holder.rs:181/) create table lock success, revision=158
b213d938-d09d-4535-8b49-5426f5cc22c9 2024-11-21T06:54:47.746668Z DEBUG databend_common_meta_client::grpc_client: [grpc_client.rs:353](http://grpc_client.rs:353/) ClientHandle([0.0.0.0:9191](http://0.0.0.0:9191/)) send request to meta client worker: request: ClientWorkerRequest { request_id: 1051, req: StreamList(Streamed(ListKVReq { prefix: "__fd_table_lock/test_tenant/135/" })) }
b213d938-d09d-4535-8b49-5426f5cc22c9 2024-11-21T06:54:47.746689Z DEBUG databend_common_meta_client::grpc_client: [grpc_client.rs:1116](http://grpc_client.rs:1116/) MetaGrpcClient([0.0.0.0:9191](http://0.0.0.0:9191/))::kv_read_v1 request: ListKV(ListKVReq { prefix: "__fd_table_lock/test_tenant/135/" })
```

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!


Discussion / hints:

Fix patch:
diff --git a/src/meta/api/src/schema_api_impl.rs b/src/meta/api/src/schema_api_impl.rs
index 6070e4cac9aac..b7320ae499f21 100644
--- a/src/meta/api/src/schema_api_impl.rs
+++ b/src/meta/api/src/schema_api_impl.rs
@@ -2738,30 +2738,49 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         &self,
         req: CreateLockRevReq,
     ) -> Result<CreateLockRevReply, KVAppError> {
-        debug!(req :? =(&req); "SchemaApi: {}", func_name!());
+        let ctx = func_name!();
+        debug!(req :? =(&req); "SchemaApi: {}", ctx);
 
         let lock_key = &req.lock_key;
+        let id_generator = IdGenerator::table_lock_id();
 
-        let revision = fetch_id(self, IdGenerator::table_lock_id()).await?;
-        let key = lock_key.gen_key(revision);
+        let mut trials = txn_backoff(None, ctx);
+        loop {
+            trials.next().unwrap()?.await;
 
-        let lock_meta = LockMeta {
-            user: req.user.clone(),
-            node: req.node.clone(),
-            query_id: req.query_id.clone(),
-            created_on: Utc::now(),
-            acquired_on: None,
-            lock_type: lock_key.lock_type(),
-            extra_info: lock_key.get_extra_info(),
-        };
+            let current_rev = self.get_seq(&id_generator).await?;
+            let revision = current_rev + 1;
+            let key = lock_key.gen_key(revision);
+            let lock_meta = LockMeta {
+                user: req.user.clone(),
+                node: req.node.clone(),
+                query_id: req.query_id.clone(),
+                created_on: Utc::now(),
+                acquired_on: None,
+                lock_type: lock_key.lock_type(),
+                extra_info: lock_key.get_extra_info(),
+            };
 
-        // Revision is unique. if it presents, consider it as success.
-        // Thus, we could just ignore create result
-        let _ = self
-            .crud_try_insert(&key, lock_meta, Some(req.ttl), || Ok::<(), Infallible>(()))
-            .await?;
+            let condition = vec![
+                txn_cond_seq(&id_generator, Eq, current_rev),
+                // assumes lock are absent.
+                txn_cond_seq(&key, Eq, 0),
+            ];
+            let if_then = vec![
+                txn_op_put(&id_generator, b"".to_vec()),
+                txn_op_put_pb(&key, &lock_meta, Some(req.ttl))?,
+            ];
+            let txn_req = TxnRequest {
+                condition,
+                if_then,
+                else_then: vec![],
+            };
+            let (succ, _responses) = send_txn(self, txn_req).await?;
 
-        Ok(CreateLockRevReply { revision })
+            if succ {
+                return Ok(CreateLockRevReply { revision });
+            }
+        }
     }
 
     #[logcall::logcall]


Test patch: