refactor: stabilize heal format recovery integration tests (#1984)

Co-authored-by: houseme <housemecn@gmail.com>
2026-08-17 10:17:55 +00:00 · 2026-02-27 15:26:19 +08:00
parent 3433dfa88e
commit 9d2b8822cf
26 changed files with 849 additions and 125 deletions
@@ -26,14 +26,19 @@ use rustfs_utils::HashAlgorithm;
 use s3s::S3ErrorCode;
 use s3s::dto::ReplicationStatus;
 use s3s::header::X_AMZ_BYPASS_GOVERNANCE_RETENTION;
+use serde::Deserialize;
 use std::fmt::Display;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 use time::OffsetDateTime;
 use tokio::sync::mpsc::{self, Receiver, Sender};

 use crate::client::utils::base64_encode;
 use crate::client::{
    api_error_response::{ErrorResponse, http_resp_to_error_response, to_error_response},
+    api_s3_datatypes::{DeleteMultiObjects, DeleteObject},
    transition_api::{ReaderImpl, RequestMetadata, TransitionClient},
 };
 use crate::{
@@ -46,6 +51,8 @@ pub struct RemoveBucketOptions {
    _forced_delete: bool,
 }

+const DELETE_RESPONSE_PREVIEW_LEN: usize = 1024;
+
 #[derive(Debug)]
 #[allow(dead_code)]
 pub struct AdvancedRemoveOptions {
@@ -354,7 +361,7 @@ impl TransitionClient {
                    body_vec.extend_from_slice(data);
                }
            }
-            process_remove_multi_objects_response(ReaderImpl::Body(Bytes::from(body_vec)), result_tx.clone());
+            process_remove_multi_objects_response(ReaderImpl::Body(Bytes::from(body_vec)), &batch, result_tx.clone()).await;
        }
        Ok(())
    }
@@ -498,11 +505,239 @@ pub struct RemoveObjectsOptions {
 }

 pub fn generate_remove_multi_objects_request(objects: &[ObjectInfo]) -> Vec<u8> {
-    todo!();
+    let escape_xml = |value: &str| -> String {
+        value
+            .replace('&', "&amp;")
+            .replace('<', "&lt;")
+            .replace('>', "&gt;")
+            .replace('\"', "&quot;")
+            .replace('\'', "&apos;")
+    };
+
+    let request: DeleteMultiObjects = DeleteMultiObjects {
+        quiet: false,
+        objects: objects
+            .iter()
+            .map(|object| DeleteObject {
+                key: object.name.clone(),
+                version_id: object.version_id.map(|v| v.to_string()).unwrap_or_default(),
+            })
+            .collect(),
+    };
+
+    match request.marshal_msg() {
+        Ok(body) => body.into_bytes(),
+        Err(_) => {
+            let mut body = String::new();
+            body.push_str("<Delete><Quiet>false</Quiet>");
+            for object in objects {
+                body.push_str("<Object>");
+                body.push_str("<Key>");
+                body.push_str(&escape_xml(&object.name));
+                body.push_str("</Key>");
+                if object.version_id.is_some() {
+                    body.push_str("<VersionId>");
+                    body.push_str(&escape_xml(&object.version_id.as_ref().map(|v| v.to_string()).unwrap_or_default()));
+                    body.push_str("</VersionId>");
+                }
+                body.push_str("</Object>");
+            }
+            body.push_str("</Delete>");
+            body.into_bytes()
+        }
+    }
 }

-pub fn process_remove_multi_objects_response(body: ReaderImpl, result_tx: Sender<RemoveObjectResult>) {
-    todo!();
+pub async fn process_remove_multi_objects_response(
+    body: ReaderImpl,
+    objects: &[ObjectInfo],
+    result_tx: Sender<RemoveObjectResult>,
+) {
+    let mut body_vec = Vec::new();
+    match body {
+        ReaderImpl::Body(content_body) => {
+            body_vec = content_body.to_vec();
+        }
+        ReaderImpl::ObjectBody(mut object_body) => match object_body.read_all().await {
+            Ok(content) => {
+                body_vec = content;
+            }
+            Err(err) => {
+                for object in objects {
+                    let version_id = object.version_id.as_ref().map(|v| v.to_string()).unwrap_or_default();
+                    let _ = result_tx
+                        .send(RemoveObjectResult {
+                            object_name: object.name.clone(),
+                            object_version_id: version_id,
+                            err: Some(std::io::Error::other(ErrorResponse {
+                                code: S3ErrorCode::Custom("ReadDeleteResponseFailed".into()),
+                                message: format!("read multi remove response failed: {err}"),
+                                bucket_name: object.bucket.clone(),
+                                key: object.name.clone(),
+                                resource: "".to_string(),
+                                request_id: "".to_string(),
+                                host_id: "".to_string(),
+                                region: "".to_string(),
+                                server: "".to_string(),
+                                status_code: StatusCode::OK,
+                            })),
+                            ..Default::default()
+                        })
+                        .await;
+                }
+                return;
+            }
+        },
+    }
+
+    #[derive(Debug, Deserialize)]
+    #[serde(rename = "DeleteResult")]
+    struct Deleted {
+        #[serde(rename = "Deleted", default)]
+        deleted: Vec<DeleteResultDeleted>,
+        #[serde(rename = "Error", default)]
+        error: Vec<DeleteResultError>,
+    }
+
+    #[derive(Debug, Deserialize)]
+    struct DeleteResultDeleted {
+        #[serde(rename = "Key")]
+        key: String,
+        #[serde(rename = "VersionId", default)]
+        version_id: String,
+        #[serde(rename = "DeleteMarker")]
+        deletemarker: bool,
+        #[serde(rename = "DeleteMarkerVersionId", default)]
+        deletemarker_version_id: String,
+    }
+
+    #[derive(Debug, Deserialize)]
+    struct DeleteResultError {
+        #[serde(rename = "Key")]
+        key: String,
+        #[serde(rename = "VersionId", default)]
+        version_id: String,
+        #[serde(rename = "Code")]
+        code: String,
+        #[serde(rename = "Message")]
+        message: String,
+    }
+
+    let mut pending = HashSet::with_capacity(objects.len());
+    for object in objects {
+        pending.insert((object.name.clone(), object.version_id.as_ref().map(|v| v.to_string()).unwrap_or_default()));
+    }
+
+    let body = String::from_utf8_lossy(&body_vec).into_owned();
+    let parsed: Deleted = match quick_xml::de::from_str(&body) {
+        Ok(parsed) => parsed,
+        Err(err) => {
+            for object in objects {
+                let version_id = object.version_id.as_ref().map(|v| v.to_string()).unwrap_or_default();
+                let _ = result_tx
+                    .send(RemoveObjectResult {
+                        object_name: object.name.clone(),
+                        object_version_id: version_id,
+                        err: Some(std::io::Error::other(ErrorResponse {
+                            code: S3ErrorCode::Custom("UnmarshalDeleteResponseFailed".into()),
+                            message: format!(
+                                "unmarshal multi remove response failed: {err}; response_body={}",
+                                body.chars().take(DELETE_RESPONSE_PREVIEW_LEN).collect::<String>()
+                            ),
+                            bucket_name: object.bucket.clone(),
+                            key: object.name.clone(),
+                            resource: "".to_string(),
+                            request_id: "".to_string(),
+                            host_id: "".to_string(),
+                            region: "".to_string(),
+                            server: "".to_string(),
+                            status_code: StatusCode::OK,
+                        })),
+                        ..Default::default()
+                    })
+                    .await;
+            }
+            return;
+        }
+    };
+
+    for deleted in parsed.deleted {
+        if !pending.remove(&(deleted.key.clone(), deleted.version_id.clone())) {
+            continue;
+        }
+
+        let _ = result_tx
+            .send(RemoveObjectResult {
+                object_name: deleted.key,
+                object_version_id: deleted.version_id,
+                delete_marker: deleted.deletemarker,
+                delete_marker_version_id: deleted.deletemarker_version_id,
+                err: None,
+            })
+            .await;
+    }
+
+    for removed in parsed.error {
+        if !pending.remove(&(removed.key.clone(), removed.version_id.clone())) {
+            continue;
+        }
+
+        let _ = result_tx
+            .send(RemoveObjectResult {
+                object_name: removed.key.clone(),
+                object_version_id: removed.version_id,
+                err: Some(std::io::Error::other(ErrorResponse {
+                    code: S3ErrorCode::Custom(removed.code.into()),
+                    message: removed.message,
+                    bucket_name: "".to_string(),
+                    key: removed.key,
+                    resource: "".to_string(),
+                    request_id: "".to_string(),
+                    host_id: "".to_string(),
+                    region: "".to_string(),
+                    server: "".to_string(),
+                    status_code: StatusCode::OK,
+                })),
+                ..Default::default()
+            })
+            .await;
+    }
+
+    for (object_name, object_version_id) in pending {
+        let bucket_name = objects
+            .iter()
+            .find(|object| {
+                object.name == object_name && object.version_id.as_ref().map(|v| v.to_string()) == Some(object_version_id.clone())
+            })
+            .map(|o| o.bucket.clone())
+            .unwrap_or_default();
+        let object_name = object_name;
+        let object_version_id = object_version_id;
+        let error_message = format!(
+            "remove response did not contain an entry for object {} with version {}",
+            object_name, object_version_id
+        );
+
+        let _ = result_tx
+            .send(RemoveObjectResult {
+                object_name: object_name.clone(),
+                object_version_id: object_version_id.clone(),
+                err: Some(std::io::Error::other(ErrorResponse {
+                    code: S3ErrorCode::Custom("UnmatchedDeleteResponseEntry".into()),
+                    message: error_message,
+                    bucket_name,
+                    key: object_name,
+                    resource: "".to_string(),
+                    request_id: "".to_string(),
+                    host_id: "".to_string(),
+                    region: "".to_string(),
+                    server: "".to_string(),
+                    status_code: StatusCode::OK,
+                })),
+                ..Default::default()
+            })
+            .await;
+    }
 }

 fn has_invalid_xml_char(str: &str) -> bool {