From 6ab14981bad9f4386f1fd9f08f2ff16b540073e7 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 27 Aug 2026 10:07:56 +0800 Subject: [PATCH] refactor(site-replication): converge send_peer_* into one request builder (#6705) --- rustfs/src/site_replication/transport.rs | 434 +++++++++++++---------- 1 file changed, 253 insertions(+), 181 deletions(-) diff --git a/rustfs/src/site_replication/transport.rs b/rustfs/src/site_replication/transport.rs index 417e75762..19138aec7 100644 --- a/rustfs/src/site_replication/transport.rs +++ b/rustfs/src/site_replication/transport.rs @@ -580,6 +580,235 @@ pub(crate) fn site_replication_peer_url(connection: &PeerConnection, wire_path: .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid site replication peer path: {e}"))) } +/// One peer admin request, collapsing the formerly-duplicated PUT/GET +/// dispatch bodies (backlog#1840 PR2): wire-path mapping, URL/authority +/// derivation, payload wire-encoding, SigV4 signing, transport-error +/// classification, and the response read live here once. The option axes are +/// the method, an optional pre-resolved client, the service-account secret +/// candidates, and the retry-event bookkeeping; the `send_peer_admin_*` +/// names below remain as thin compatibility wrappers. +pub(crate) struct PeerAdminRequest<'a> { + connection: &'a PeerConnection, + path: &'a str, + access_key: &'a str, + method: Method, + client: Option<&'a reqwest::Client>, +} + +impl<'a> PeerAdminRequest<'a> { + pub(crate) fn put(connection: &'a PeerConnection, path: &'a str, access_key: &'a str) -> Self { + Self { + connection, + path, + access_key, + method: Method::PUT, + client: None, + } + } + + pub(crate) fn get(connection: &'a PeerConnection, path: &'a str, access_key: &'a str) -> Self { + Self { + method: Method::GET, + ..Self::put(connection, path, access_key) + } + } + + pub(crate) fn with_client(mut self, client: &'a reqwest::Client) -> Self { + self.client = Some(client); + self + } + + async fn resolved_client(&self) -> S3Result { + match self.client { + Some(client) => Ok(client.clone()), + None => site_replication_client_for(self.connection).await, + } + } + + /// Send and return the raw status/body without a success check. `body` is + /// the JSON payload of a `PUT` (serialized and wire-encoded here, which + /// is where the peer-join encryption applies); `None` sends a bodiless + /// request (the `GET` flavor). + pub(crate) async fn send_raw(&self, secret_key: &str, body: Option<&T>) -> S3Result<(StatusCode, Vec)> { + let client = self.resolved_client().await?; + let path = site_replication_peer_wire_path(self.path); + let url = site_replication_peer_url(self.connection, &path)?; + let uri = url + .as_str() + .parse::() + .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid peer endpoint: {e}")))?; + let authority = uri + .authority() + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "peer endpoint missing authority".to_string()))? + .to_string(); + let payload = body + .map(|body| { + let payload = serde_json::to_vec(body).map_err(|e| { + S3Error::with_message(S3ErrorCode::InternalError, format!("serialize peer request failed: {e}")) + })?; + site_replication_peer_payload(&path, secret_key, payload) + }) + .transpose()?; + + let mut request = http::Request::builder() + .method(self.method.clone()) + .uri(uri) + .header(HOST, authority) + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if let Some((_, content_type)) = &payload { + request = request.header(CONTENT_TYPE, *content_type); + } + let signed = sign_v4( + request + .body(Body::empty()) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("build peer request failed: {e}")))?, + payload.as_ref().map(|(payload, _)| payload.len() as i64).unwrap_or(0), + self.access_key, + secret_key, + "", + current_region() + .map(|region| region.to_string()) + .as_deref() + .unwrap_or("us-east-1"), + ); + + let mut req = client.request(self.method.clone(), url.clone()); + for (name, value) in signed.headers() { + req = req.header(name, value); + } + if let Some((payload, _)) = payload { + req = req.body(payload); + } + + let response = req.send().await.map_err(|e| { + let classify = if e.is_timeout() { + "timeout" + } else if e.is_connect() && e.to_string().to_ascii_lowercase().contains("dns") { + "dns resolution" + } else if e.to_string().to_ascii_lowercase().contains("certificate") + || e.to_string().to_ascii_lowercase().contains("tls") + { + "tls handshake" + } else if e.is_connect() { + "connect" + } else { + "request" + }; + S3Error::with_message(S3ErrorCode::InternalError, format!("peer request to {url} failed ({classify}): {e}")) + })?; + + let status = response.status(); + let body = response + .bytes() + .await + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("read peer response failed: {e}")))?; + + Ok((status, body.to_vec())) + } + + /// `PUT` with the success check and error shape of + /// [`send_peer_admin_request`]. + pub(crate) async fn send(&self, secret_key: &str, body: &T) -> S3Result> { + let (status, body) = self.send_raw(secret_key, Some(body)).await?; + if status.is_success() { + return Ok(body); + } + + let detail = String::from_utf8_lossy(&body).into_owned(); + Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!( + "peer request to {}{} failed with {status}: {detail}", + self.connection.endpoint(), + self.path + ), + )) + } + + /// Bodiless `GET` with the success check and error shape of + /// [`send_peer_admin_get_request`]. + pub(crate) async fn send_get(&self, secret_key: &str) -> S3Result> { + let (status, body) = self.send_raw::<()>(secret_key, None).await?; + if !status.is_success() { + let url = site_replication_peer_url(self.connection, &site_replication_peer_wire_path(self.path))?; + let detail = String::from_utf8_lossy(&body).into_owned(); + return Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!("peer request to {url} failed with {status}: {detail}"), + )); + } + + Ok(body) + } + + /// Try each distinct non-empty service-account secret until one is + /// accepted; stop early on an error that cannot be a secret mismatch. + pub(crate) async fn send_with_secret_candidates( + &self, + secret_candidates: &[String], + body: &T, + ) -> S3Result> { + let client = self.resolved_client().await?; + let request = PeerAdminRequest { + connection: self.connection, + path: self.path, + access_key: self.access_key, + method: self.method.clone(), + client: Some(&client), + }; + let mut tried = HashSet::new(); + let mut errors = Vec::new(); + + for secret_key in secret_candidates.iter().filter(|secret_key| !secret_key.is_empty()) { + if !tried.insert(secret_key.as_str()) { + continue; + } + + match request.send(secret_key, body).await { + Ok(body) => return Ok(body), + Err(err) => { + let detail = format!("{err}"); + let may_retry_with_next_secret = peer_error_may_be_secret_mismatch(&detail); + errors.push(summarize_peer_error_detail(&detail)); + if !may_retry_with_next_secret { + break; + } + } + } + } + + Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!( + "peer request to {}{} failed with all service-account secrets: {}", + self.connection.endpoint(), + self.path, + errors.join("; ") + ), + )) + } + + /// [`Self::send`] plus the retry-queue bookkeeping: a success settles the + /// peer/path's queued event, a failure enqueues one. + pub(crate) async fn send_with_retry_event( + &self, + peer: &PeerInfo, + secret_key: &str, + body: &T, + ) -> S3Result> { + match self.send(secret_key, body).await { + Ok(body) => { + dequeue_site_replication_retry_event(peer, self.path).await; + Ok(body) + } + Err(err) => { + enqueue_site_replication_retry_event(peer, self.path, &err).await; + Err(err) + } + } + } +} + #[cfg(test)] pub(crate) async fn send_peer_admin_request_raw( connection: &PeerConnection, @@ -588,8 +817,9 @@ pub(crate) async fn send_peer_admin_request_raw( secret_key: &str, body: &T, ) -> S3Result<(StatusCode, Vec)> { - let client = site_replication_client_for(connection).await?; - send_peer_admin_request_raw_with_client(&client, connection, path, access_key, secret_key, body).await + PeerAdminRequest::put(connection, path, access_key) + .send_raw(secret_key, Some(body)) + .await } pub(crate) async fn send_peer_admin_request_raw_with_client( @@ -600,67 +830,10 @@ pub(crate) async fn send_peer_admin_request_raw_with_client( secret_key: &str, body: &T, ) -> S3Result<(StatusCode, Vec)> { - let path = site_replication_peer_wire_path(path); - let url = site_replication_peer_url(connection, &path)?; - let uri = url - .as_str() - .parse::() - .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid peer endpoint: {e}")))?; - let authority = uri - .authority() - .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "peer endpoint missing authority".to_string()))? - .to_string(); - let payload = serde_json::to_vec(body) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize peer request failed: {e}")))?; - let (payload, content_type) = site_replication_peer_payload(&path, secret_key, payload)?; - - let signed = sign_v4( - http::Request::builder() - .method(Method::PUT) - .uri(uri) - .header(HOST, authority) - .header("x-amz-content-sha256", UNSIGNED_PAYLOAD) - .header(CONTENT_TYPE, content_type) - .body(Body::empty()) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("build peer request failed: {e}")))?, - payload.len() as i64, - access_key, - secret_key, - "", - current_region() - .map(|region| region.to_string()) - .as_deref() - .unwrap_or("us-east-1"), - ); - - let mut req = client.request(reqwest::Method::PUT, url.clone()); - for (name, value) in signed.headers() { - req = req.header(name, value); - } - - let response = req.body(payload).send().await.map_err(|e| { - let classify = if e.is_timeout() { - "timeout" - } else if e.is_connect() && e.to_string().to_ascii_lowercase().contains("dns") { - "dns resolution" - } else if e.to_string().to_ascii_lowercase().contains("certificate") || e.to_string().to_ascii_lowercase().contains("tls") - { - "tls handshake" - } else if e.is_connect() { - "connect" - } else { - "request" - }; - S3Error::with_message(S3ErrorCode::InternalError, format!("peer request to {url} failed ({classify}): {e}")) - })?; - - let status = response.status(); - let body = response - .bytes() + PeerAdminRequest::put(connection, path, access_key) + .with_client(client) + .send_raw(secret_key, Some(body)) .await - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("read peer response failed: {e}")))?; - - Ok((status, body.to_vec())) } pub(crate) async fn send_peer_admin_request( @@ -670,8 +843,9 @@ pub(crate) async fn send_peer_admin_request( secret_key: &str, body: &T, ) -> S3Result> { - let client = site_replication_client_for(connection).await?; - send_peer_admin_request_with_client(&client, connection, path, access_key, secret_key, body).await + PeerAdminRequest::put(connection, path, access_key) + .send(secret_key, body) + .await } pub(crate) async fn send_peer_admin_request_with_client( @@ -682,16 +856,10 @@ pub(crate) async fn send_peer_admin_request_with_client( secret_key: &str, body: &T, ) -> S3Result> { - let (status, body) = send_peer_admin_request_raw_with_client(client, connection, path, access_key, secret_key, body).await?; - if status.is_success() { - return Ok(body); - } - - let detail = String::from_utf8_lossy(&body).into_owned(); - Err(S3Error::with_message( - S3ErrorCode::InternalError, - format!("peer request to {}{path} failed with {status}: {detail}", connection.endpoint()), - )) + PeerAdminRequest::put(connection, path, access_key) + .with_client(client) + .send(secret_key, body) + .await } pub(crate) async fn send_peer_admin_request_with_secret_candidates( @@ -701,36 +869,9 @@ pub(crate) async fn send_peer_admin_request_with_secret_candidates secret_candidates: &[String], body: &T, ) -> S3Result> { - let client = site_replication_client_for(connection).await?; - let mut tried = HashSet::new(); - let mut errors = Vec::new(); - - for secret_key in secret_candidates.iter().filter(|secret_key| !secret_key.is_empty()) { - if !tried.insert(secret_key.as_str()) { - continue; - } - - match send_peer_admin_request_with_client(&client, connection, path, access_key, secret_key, body).await { - Ok(body) => return Ok(body), - Err(err) => { - let detail = format!("{err}"); - let may_retry_with_next_secret = peer_error_may_be_secret_mismatch(&detail); - errors.push(summarize_peer_error_detail(&detail)); - if !may_retry_with_next_secret { - break; - } - } - } - } - - Err(S3Error::with_message( - S3ErrorCode::InternalError, - format!( - "peer request to {}{path} failed with all service-account secrets: {}", - connection.endpoint(), - errors.join("; ") - ), - )) + PeerAdminRequest::put(connection, path, access_key) + .send_with_secret_candidates(secret_candidates, body) + .await } pub(crate) fn peer_error_may_be_secret_mismatch(detail: &str) -> bool { @@ -748,8 +889,7 @@ pub(crate) async fn send_peer_admin_get_request( access_key: &str, secret_key: &str, ) -> S3Result> { - let client = site_replication_client_for(connection).await?; - send_peer_admin_get_request_with_client(&client, connection, path, access_key, secret_key).await + PeerAdminRequest::get(connection, path, access_key).send_get(secret_key).await } pub(crate) async fn send_peer_admin_get_request_with_client( @@ -759,71 +899,10 @@ pub(crate) async fn send_peer_admin_get_request_with_client( access_key: &str, secret_key: &str, ) -> S3Result> { - let path = site_replication_peer_wire_path(path); - let url = site_replication_peer_url(connection, &path)?; - let uri = url - .as_str() - .parse::() - .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid peer endpoint: {e}")))?; - let authority = uri - .authority() - .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "peer endpoint missing authority".to_string()))? - .to_string(); - - let signed = sign_v4( - http::Request::builder() - .method(Method::GET) - .uri(uri) - .header(HOST, authority) - .header("x-amz-content-sha256", UNSIGNED_PAYLOAD) - .body(Body::empty()) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("build peer request failed: {e}")))?, - 0, - access_key, - secret_key, - "", - current_region() - .map(|region| region.to_string()) - .as_deref() - .unwrap_or("us-east-1"), - ); - - let mut req = client.request(reqwest::Method::GET, url.clone()); - for (name, value) in signed.headers() { - req = req.header(name, value); - } - - let response = req.send().await.map_err(|e| { - let classify = if e.is_timeout() { - "timeout" - } else if e.is_connect() && e.to_string().to_ascii_lowercase().contains("dns") { - "dns resolution" - } else if e.to_string().to_ascii_lowercase().contains("certificate") || e.to_string().to_ascii_lowercase().contains("tls") - { - "tls handshake" - } else if e.is_connect() { - "connect" - } else { - "request" - }; - S3Error::with_message(S3ErrorCode::InternalError, format!("peer request to {url} failed ({classify}): {e}")) - })?; - - let status = response.status(); - let body = response - .bytes() + PeerAdminRequest::get(connection, path, access_key) + .with_client(client) + .send_get(secret_key) .await - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("read peer response failed: {e}")))?; - - if !status.is_success() { - let detail = String::from_utf8_lossy(&body).into_owned(); - return Err(S3Error::with_message( - S3ErrorCode::InternalError, - format!("peer request to {url} failed with {status}: {detail}"), - )); - } - - Ok(body.to_vec()) } pub(crate) async fn runtime_site_replication_targets() -> S3Result> { @@ -909,17 +988,10 @@ pub(crate) async fn send_peer_admin_request_with_retry_event_transport S3Result> { - match send_peer_admin_request_with_client(&transport.client, &transport.connection, path, access_key, secret_key, body).await - { - Ok(body) => { - dequeue_site_replication_retry_event(peer, path).await; - Ok(body) - } - Err(err) => { - enqueue_site_replication_retry_event(peer, path, &err).await; - Err(err) - } - } + PeerAdminRequest::put(&transport.connection, path, access_key) + .with_client(&transport.client) + .send_with_retry_event(peer, secret_key, body) + .await } pub(crate) fn parse_endpoint_refresh_status(peer: &PeerInfo, body: &[u8]) -> S3Result<()> {