fix: Handle concurrent gc_dropped_db_by_id and undrop_database operations safely (#17901)

* fix: Handle concurrent `gc_dropped_db_by_id` and `undrop_database` operations safely

Add transaction condition to ensure proper behavior when `gc_dropped_db_by_id` and
`undrop_database` are executed concurrently on the same database, by checking that
that database meta being cleaned is marked as dropped.

* tweak comments

* tweak UT

Author: dantengsky

src/meta/api/src/schema_api_impl.rs

@@ -3725,6 +3725,14 @@ async fn gc_dropped_db_by_id(
         return Ok(());
     };
 
+    if seq_db_meta.drop_on.is_none() {
+        // If db is not marked as dropped, just ignore the gc request and return directly.
+        // In subsequent KV transactions, we also verify that db_meta hasn't changed
+        // to ensure we don't reclaim metadata of the given database that might have been
+        // successfully undropped in a parallel operation.
+        return Ok(());
+    }
+
     // TODO: enable this when gc_in_progress is set.
     // if !seq_db_meta.gc_in_progress {
     //     let err = UnknownDatabaseId::new(
@@ -3778,6 +3786,9 @@ async fn gc_dropped_db_by_id(
             .push(txn_put_pb(&db_id_history_ident, &db_id_list)?);
     }
 
+    // Verify db_meta hasn't changed since we started this operation.
+    // This establishes a condition for the transaction that will prevent it from committing
+    // if the database metadata was modified by another concurrent operation (like un-dropping).
     txn.condition.push(txn_cond_eq_seq(&dbid, seq_db_meta.seq));
     txn.if_then.push(txn_op_del(&dbid));
     txn.condition

src/meta/api/src/schema_api_test_suite.rs

@@ -337,6 +337,7 @@ impl SchemaApiTestSuite {
         suite.table_index_create_drop(&b.build().await).await?;
         suite.index_create_list_drop(&b.build().await).await?;
         suite.table_lock_revision(&b.build().await).await?;
+        suite.gc_dropped_db_after_undrop(&b.build().await).await?;
         suite.catalog_create_get_list_drop(&b.build().await).await?;
         suite.table_least_visible_time(&b.build().await).await?;
         suite
@@ -6946,6 +6947,79 @@ impl SchemaApiTestSuite {
         Ok(())
     }
 
+    async fn gc_dropped_db_after_undrop<MT: SchemaApi + kvapi::AsKVApi<Error = MetaError>>(
+        self,
+        mt: &MT,
+    ) -> anyhow::Result<()> {
+        let tenant_name = "tenant1_gc_dropped_db_after_undrop";
+        let db_name = "db1_gc_dropped_db_after_undrop";
+        let mut util = Util::new(mt, tenant_name, db_name, "", "eng");
+
+        let tenant = Tenant::new_or_err(tenant_name, func_name!())?;
+        let db_name_ident = DatabaseNameIdent::new(&tenant, db_name);
+
+        // 1. Create database
+        util.create_db().await?;
+        let db_id = util.db_id;
+
+        info!("Created database with ID: {}", db_id);
+
+        // 2. Drop database
+        util.drop_db().await?;
+
+        // 2.1. Check database is marked as dropped
+        let req = ListDroppedTableReq::new(&tenant);
+        let resp = mt.get_drop_table_infos(req).await?;
+
+        // Filter for our specific database ID
+        let drop_ids: Vec<DroppedId> = resp
+            .drop_ids
+            .into_iter()
+            .filter(|id| {
+                if let DroppedId::Db { db_id: id, .. } = id {
+                    *id == db_id
+                } else {
+                    false
+                }
+            })
+            .collect();
+
+        assert!(
+            !drop_ids.is_empty(),
+            "Database being tested should be dropped"
+        );
+
+        // 3. Undrop the database
+        //
+        // A more rigorous test would verify the race condition protection, but difficult to implement as an integration test:
+        // Ideally, we would undrop the database precisely after the `gc_drop_tables` process has verified
+        // that the database is marked as dropped, but before committing the kv transaction that removes the database metadata.
+
+        let undrop_req = UndropDatabaseReq {
+            name_ident: db_name_ident.clone(),
+        };
+        mt.undrop_database(undrop_req).await?;
+
+        let req = GcDroppedTableReq {
+            tenant: tenant.clone(),
+            drop_ids,
+        };
+
+        // 4. Check that gc_drop_tables operation has NOT removed database's meta data
+        mt.gc_drop_tables(req.clone()).await?;
+
+        // 5. Verify the database is still accessible
+        let get_req = GetDatabaseReq::new(tenant.clone(), db_name.to_string());
+        let db_info = mt.get_database(get_req).await?;
+        assert_eq!(db_info.database_id.db_id, db_id, "Database ID should match");
+        assert!(
+            db_info.meta.drop_on.is_none(),
+            "Database should not be marked as dropped"
+        );
+
+        Ok(())
+    }
+
     // pub async fn share_create_get_drop<MT: SchemaApi>(&self, mt: &MT) -> anyhow::Result<()> {
     //     let tenant1 = "tenant1";
     //     let share_name1 = "share1";