refactor(meta): minor structural cleanups in process, control, admin, semaphore (#19470)

Fix 5 code quality items identified in the meta crate review:

- process/kv_processor.rs: replace `unreachable!()` with `debug_assert!()` to
  avoid panics in release builds; `unreachable!` should be used only when the
  compiler cannot prove a code path is impossible.

- process/process_meta_dir.rs: move the unused `v1_ent` suppression into the
  destructuring pattern (`_v1_ent`) instead of a separate `let _ = v1_ent`
  binding.

- control/export_from_grpc.rs: replace `is_some() + unwrap()` with
  `if let Some(ref mut f)`, eliminating redundant borrows.

- plugins/semaphore/semaphore.rs: initialize `seq_policy` directly with
  `GeneratorKey` instead of first setting `TimeBased` then overwriting it.

- admin/v1/ctrl.rs: add a local `internal_err` helper (matching the one in
  `features.rs`) and replace the three verbose
  `map_err(|e| poem::Error::from_string(e.to_string(), INTERNAL_SERVER_ERROR))`
  chains with `.map_err(internal_err)`.

Author: drmingdrmer

src/meta/admin/src/v1/ctrl.rs

@@ -44,12 +44,8 @@ impl<SP: SpawnApi> HttpService<SP> {
         meta_handle
             .handle_trigger_snapshot()
             .await
-            .map_err(|e| {
-                poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
-            })?
-            .map_err(|e| {
-                poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
-            })?;
+            .map_err(internal_err)?
+            .map_err(internal_err)?;
         Ok(Json(()).into_response())
     }
 
@@ -60,9 +56,7 @@ impl<SP: SpawnApi> HttpService<SP> {
         let metrics = meta_handle
             .handle_raft_metrics()
             .await
-            .map_err(|e| {
-                poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
-            })?
+            .map_err(internal_err)?
             .borrow_watched()
             .clone();
 
@@ -116,12 +110,8 @@ impl<SP: SpawnApi> HttpService<SP> {
         meta_handle
             .handle_trigger_transfer_leader(to)
             .await
-            .map_err(|e| {
-                poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
-            })?
-            .map_err(|e| {
-                poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
-            })?;
+            .map_err(internal_err)?
+            .map_err(internal_err)?;
 
         Ok(Json(TransferLeaderResponse {
             from: id,
@@ -131,3 +121,7 @@ impl<SP: SpawnApi> HttpService<SP> {
         .into_response())
     }
 }
+
+fn internal_err(e: impl std::fmt::Display) -> poem::Error {
+    poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
+}

src/meta/control/src/export_from_grpc.rs

@@ -91,7 +91,7 @@ pub async fn export_from_grpc(
 
     let mut stream = exported.into_inner();
 
-    let file: Option<File> = if !save.is_empty() {
+    let mut file: Option<File> = if !save.is_empty() {
         eprintln!("    To:   File: {}", save);
         Some(File::create(&save)?)
     } else {
@@ -114,18 +114,16 @@ pub async fn export_from_grpc(
                 }
             }
 
-            if file.as_ref().is_none() {
-                println!("{}", line);
+            if let Some(ref mut f) = file {
+                f.write_all(format!("{}\n", line).as_bytes())?;
             } else {
-                file.as_ref()
-                    .unwrap()
-                    .write_all(format!("{}\n", line).as_bytes())?;
+                println!("{}", line);
             }
         }
     }
 
-    if file.as_ref().is_some() {
-        file.as_ref().unwrap().sync_all()?;
+    if let Some(ref f) = file {
+        f.sync_all()?;
     }
 
     Ok(())

src/meta/plugins/semaphore/src/semaphore.rs

@@ -164,10 +164,10 @@ impl Semaphore {
         let uniq = UNIQ.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
 
         let mut sem = Semaphore {
-            prefix,
-            seq_policy: SeqPolicy::TimeBased {
-                timestamp: SystemTime::now().duration_since(UNIX_EPOCH).unwrap(),
+            seq_policy: SeqPolicy::GeneratorKey {
+                generator_key: Self::seq_generator_key_for(&prefix),
             },
+            prefix,
             meta_client,
             permit_ttl,
             subscriber_task_handle: None,
@@ -176,11 +176,6 @@ impl Semaphore {
             uniq,
         };
 
-        // The default is using generator key
-        sem.seq_policy = SeqPolicy::GeneratorKey {
-            generator_key: sem.seq_generator_key(),
-        };
-
         sem.spawn_meta_event_subscriber(tx, capacity, cancel_rx)
             .await;
 
@@ -326,7 +321,11 @@ impl Semaphore {
     ///
     /// Update this key in the meta-service to obtain a new sequence number.
     fn seq_generator_key(&self) -> String {
-        format!("{}/seq_generator", self.prefix)
+        Self::seq_generator_key_for(&self.prefix)
+    }
+
+    fn seq_generator_key_for(prefix: &str) -> String {
+        format!("{}/seq_generator", prefix)
     }
 
     /// The left-close right-open range for the semaphore ids.

src/meta/process/src/kv_processor.rs

@@ -187,9 +187,10 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
     }
 
     fn proc_condition(&self, c: TxnCondition) -> TxnCondition {
-        if let Some(Target::Value(_)) = &c.target {
-            unreachable!("we has never used value");
-        }
+        debug_assert!(
+            !matches!(&c.target, Some(Target::Value(_))),
+            "we have never used value"
+        );
         c
     }
 

src/meta/process/src/process_meta_dir.rs

@@ -51,8 +51,7 @@ where F: Fn(RaftStoreEntry) -> Result<Option<RaftStoreEntry>, anyhow::Error> {
             converted.len()
         );
 
-        for (v1_ent, v2_ent) in converted {
-            let _ = v1_ent;
+        for (_v1_ent, v2_ent) in converted {
             let (k, v) = RaftStoreEntry::serialize(&v2_ent)?;
             tree.insert(k, v)?;
         }