fix: resolve Rust 1.99 Clippy warnings (#8301)

Replace deprecated atomic fetch_update calls with try_update, remove
redundant closure borrows, and drop unnecessary async-recursion macros.
Keep atomic orderings and explicitly boxed recursive calls unchanged.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/cluster/rpc/network_probe.rs

@@ -56,7 +56,7 @@ impl DiagnosticPacing {
         // Reservation is never refunded: a failed or dropped RPC may have sent some payload.
         let previous = self
             .charged_bytes
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |charged| {
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |charged| {
                 charged.checked_add(bytes).filter(|total| *total <= MAX_NETWORK_PROBE_BYTES)
             })
             .map_err(|_| NetworkPeerProbeError::LimitExceeded)?;

crates/ecstore/src/core/pools.rs

@@ -12976,7 +12976,7 @@ impl ECStore {
                 .targets
                 .iter()
                 .find(|target| target.pool_index == permitted_target_pool_index)
-                .and_then(&candidate)
+                .and_then(candidate)
             {
                 return Ok(pool_index);
             }

crates/ecstore/src/data_movement/scanner_backlog.rs

@@ -306,7 +306,7 @@ pub mod test_util {
         let Some(fault) = fault else { return Ok(None) };
         if fault
             .remaining
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1))
             != Ok(1)
         {
             return Ok(None);

crates/ecstore/src/disk/local.rs

@@ -6798,7 +6798,6 @@ impl LocalDisk {
     }
 
     #[tracing::instrument(name = "delete_file", level = "trace", skip_all)]
-    #[async_recursion::async_recursion]
     async fn delete_file_with_namespace_owner(
         &self,
         base_path: &PathBuf,
@@ -7952,7 +7951,6 @@ impl LocalDisk {
         Err(DiskError::FileCorrupt)
     }
 
-    #[async_recursion::async_recursion]
     #[allow(clippy::too_many_arguments)]
     async fn scan_dir<W>(
         &self,

crates/ecstore/src/disk/uring_result_budget.rs

@@ -73,7 +73,7 @@ impl ProcessResultBudget {
         }
         inner
             .available
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |available| available.checked_sub(bytes))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |available| available.checked_sub(bytes))
             .map(|_| {
                 Some(ResultBudgetReservation {
                     inner: Arc::clone(inner),

crates/ecstore/src/layout/endpoints.rs

@@ -908,7 +908,7 @@ fn explicit_url_port(raw: &str) -> Result<Option<u16>> {
         return Err(invalid());
     }
 
-    let (scheme, remainder) = raw.split_once("://").ok_or_else(&invalid)?;
+    let (scheme, remainder) = raw.split_once("://").ok_or_else(invalid)?;
     if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
         return Err(invalid());
     }
@@ -916,17 +916,17 @@ fn explicit_url_port(raw: &str) -> Result<Option<u16>> {
         .split('/')
         .next()
         .filter(|authority| !authority.is_empty())
-        .ok_or_else(&invalid)?;
+        .ok_or_else(invalid)?;
     if authority.contains('@') {
         return Err(invalid());
     }
 
     let port = if let Some(bracketed) = authority.strip_prefix('[') {
-        let (_, suffix) = bracketed.split_once(']').ok_or_else(&invalid)?;
+        let (_, suffix) = bracketed.split_once(']').ok_or_else(invalid)?;
         if suffix.is_empty() {
             return Ok(None);
         }
-        suffix.strip_prefix(':').ok_or_else(&invalid)?
+        suffix.strip_prefix(':').ok_or_else(invalid)?
     } else if let Some((_, port)) = authority.rsplit_once(':') {
         port
     } else {
@@ -1001,11 +1001,7 @@ fn parse_explicit_local_endpoint_host(raw: &str) -> Result<Host<String>> {
 
     let host = Host::parse(raw).map_err(|_| invalid())?;
     let host = match host {
-        Host::Domain(domain) => Host::Domain(
-            domain_without_optional_trailing_dot(&domain)
-                .ok_or_else(&invalid)?
-                .to_string(),
-        ),
+        Host::Domain(domain) => Host::Domain(domain_without_optional_trailing_dot(&domain).ok_or_else(invalid)?.to_string()),
         host => host,
     };
     if matches!(&host, Host::Domain(domain) if domain.contains('*'))

crates/ecstore/src/runtime/instance.rs

@@ -396,7 +396,7 @@ impl InstanceContext {
     pub(crate) fn begin_namespace_commit(self: &Arc<Self>) -> Arc<NamespaceCommitGuard> {
         let counted = self
             .namespace_commits
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
             .is_ok();
         if counted {
             self.advance_namespace_commit_generation();
@@ -412,7 +412,7 @@ impl InstanceContext {
     fn advance_namespace_commit_generation(&self) {
         let _ = self
             .namespace_commit_generation
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| Some(generation.saturating_add(1)));
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |generation| Some(generation.saturating_add(1)));
     }
 
     pub(crate) fn namespace_commit_generation(&self) -> u64 {

crates/ecstore/src/store/init.rs

@@ -4362,7 +4362,7 @@ mod tests {
             // successful attempts. Only injected faults spend this global budget.
             // A real failure may consume an attempt, so preserve the final chance.
             faults
-                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
+                .try_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
                     (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1))
                         .then_some(faults.saturating_add(1))
                 })

crates/s3select-query/src/sql/planner.rs

@@ -14,7 +14,6 @@
 
 use std::{convert::Infallible, ops::ControlFlow};
 
-use async_recursion::async_recursion;
 use async_trait::async_trait;
 use datafusion::sql::{
     planner::{IdentNormalizer, SqlToRel},
@@ -58,7 +57,6 @@ impl<'a, S: ContextProviderExtension + Send + Sync + 'a> SqlPlanner<'a, S> {
     }
 
     /// Generate a logical plan from an  Extent SQL statement
-    #[async_recursion]
     pub(crate) async fn statement_to_plan(&self, statement: ExtStatement, session: &SessionCtx) -> QueryResult<Plan> {
         match statement {
             ExtStatement::SqlStatement(stmt) => self.df_sql_to_plan(*stmt, session).await,