From 791af1888fe35ec08007599cf809ddbde7255d42 Mon Sep 17 00:00:00 2001 From: Chris Date: Fri, 2 Oct 2026 05:43:24 +0800 Subject: [PATCH 1/2] fix(connect): preserve site replication S3 key path (#8300) --- .../diagnostics/perf_site_replication.rs | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/rustfs/src/connect/diagnostics/perf_site_replication.rs b/rustfs/src/connect/diagnostics/perf_site_replication.rs index 66fd64d58..90907cc1c 100644 --- a/rustfs/src/connect/diagnostics/perf_site_replication.rs +++ b/rustfs/src/connect/diagnostics/perf_site_replication.rs @@ -33,7 +33,6 @@ use bytes::Bytes; use futures::StreamExt as _; use p256::ecdsa::{Signature, SigningKey, signature::Signer as _}; use p256::pkcs8::DecodePrivateKey as _; -use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode}; use reqwest::{Client, Method, Response, StatusCode, Url}; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; @@ -1317,7 +1316,7 @@ async fn drain_response( fn object_url(endpoint: &Url, bucket: &str, key: &str, version_id: Option<&str>) -> Result { let mut url = endpoint - .join(&format!("{bucket}/{}", encode_path(key))) + .join(&format!("{bucket}/{key}")) .map_err(|_| SiteReplicationProbeError::ProtocolFailure)?; if let Some(version_id) = version_id { url.query_pairs_mut().append_pair("versionId", version_id); @@ -1333,14 +1332,6 @@ fn list_versions_url(endpoint: &Url, bucket: &str, key: &str) -> Result String { - value - .split('/') - .map(|segment| utf8_percent_encode(segment, NON_ALPHANUMERIC).to_string()) - .collect::>() - .join("/") -} - fn deployment_endpoint(value: &str) -> Result { let mut url = Url::parse(value).map_err(|_| SiteReplicationPerformanceError::InvalidEndpoint)?; let local_http = url.scheme() == "http" @@ -1655,6 +1646,12 @@ mod tests { #[test] fn cleanup_queries_are_version_specific_and_task_scoped() { let endpoint = Url::parse("https://source.example/").expect("endpoint"); + let scoped = object_url(&endpoint, "scratch-bucket", "rustfs-connect/site-replication/019c-1234", None) + .expect("scoped object URL"); + assert_eq!( + scoped.as_str(), + "https://source.example/scratch-bucket/rustfs-connect/site-replication/019c-1234" + ); let object = object_url(&endpoint, "scratch-bucket", "path/a b", Some("version+1")).expect("object URL"); assert_eq!(object.as_str(), "https://source.example/scratch-bucket/path/a%20b?versionId=version%2B1"); let list = list_versions_url(&endpoint, "scratch-bucket", "path/a b").expect("list URL"); From 370b517b4fa526a381ecbdf5e74088d40fbf0ce2 Mon Sep 17 00:00:00 2001 From: Hauser Date: Fri, 2 Oct 2026 10:11:22 +0800 Subject: [PATCH 2/2] 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 Co-authored-by: zhi22915 --- crates/ecstore/src/cluster/rpc/network_probe.rs | 2 +- crates/ecstore/src/core/pools.rs | 2 +- .../ecstore/src/data_movement/scanner_backlog.rs | 2 +- crates/ecstore/src/disk/local.rs | 2 -- crates/ecstore/src/disk/uring_result_budget.rs | 2 +- crates/ecstore/src/layout/endpoints.rs | 14 +++++--------- crates/ecstore/src/runtime/instance.rs | 4 ++-- crates/ecstore/src/store/init.rs | 2 +- crates/s3select-query/src/sql/planner.rs | 2 -- 9 files changed, 12 insertions(+), 20 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/network_probe.rs b/crates/ecstore/src/cluster/rpc/network_probe.rs index b4e58baad..75d9049ea 100644 --- a/crates/ecstore/src/cluster/rpc/network_probe.rs +++ b/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)?; diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index b40b520ff..66d21ff25 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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); } diff --git a/crates/ecstore/src/data_movement/scanner_backlog.rs b/crates/ecstore/src/data_movement/scanner_backlog.rs index c00dec987..22c4a925c 100644 --- a/crates/ecstore/src/data_movement/scanner_backlog.rs +++ b/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); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1c62872f3..2bf70a791 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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( &self, diff --git a/crates/ecstore/src/disk/uring_result_budget.rs b/crates/ecstore/src/disk/uring_result_budget.rs index fa1d844c5..b00959325 100644 --- a/crates/ecstore/src/disk/uring_result_budget.rs +++ b/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), diff --git a/crates/ecstore/src/layout/endpoints.rs b/crates/ecstore/src/layout/endpoints.rs index 1d9ad5faa..d6f17d9b3 100644 --- a/crates/ecstore/src/layout/endpoints.rs +++ b/crates/ecstore/src/layout/endpoints.rs @@ -908,7 +908,7 @@ fn explicit_url_port(raw: &str) -> Result> { 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> { .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> { 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('*')) diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 16a974af7..7e187c3a0 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -396,7 +396,7 @@ impl InstanceContext { pub(crate) fn begin_namespace_commit(self: &Arc) -> Arc { 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 { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a18bffe04..c7654c1ac 100644 --- a/crates/ecstore/src/store/init.rs +++ b/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)) }) diff --git a/crates/s3select-query/src/sql/planner.rs b/crates/s3select-query/src/sql/planner.rs index b847c07d9..57feec67c 100644 --- a/crates/s3select-query/src/sql/planner.rs +++ b/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 { match statement { ExtStatement::SqlStatement(stmt) => self.df_sql_to_plan(*stmt, session).await,