Fix S3 versioning (live-object VersionId, DM PUT/DELETE), harden DeleteObjects/ListObjects conformance, and run hot paths on blocking threads

- Shallow listing: read per-directory _index.json once for eTags instead
  of N serial .meta.json reads. Validate prefix for path traversal and
  verify normalized target stays within bucket root.
- Recursive listing: cache full per-directory index during the walk so
  each _index.json is parsed at most once per call.
- Per-bucket listing cache with 5s TTL and per-bucket rebuild mutex.
  Invalidated on put/delete/copy/metadata/tags/multipart-complete.
  Pagination uses partition_point for O(log n) start lookup.
- UI stream endpoint now actually streams via mpsc + Body::from_stream
  instead of buffering into a Vec<String>. Cancels producer on client
  disconnect.
- UI JSON endpoint honors delimiter=/ and returns common_prefixes.
- run_blocking wrapper dispatches sync filesystem work via
  block_in_place on multi-threaded runtimes, falls back to inline on
  current-thread runtimes (unit tests).

Cargo.lock

@@ -2639,7 +2639,7 @@ dependencies = [
 
 [[package]]
 name = "myfsio-auth"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "aes",
  "base64",
@@ -2664,7 +2664,7 @@ dependencies = [
 
 [[package]]
 name = "myfsio-common"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "chrono",
  "serde",
@@ -2675,7 +2675,7 @@ dependencies = [
 
 [[package]]
 name = "myfsio-crypto"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "aes-gcm",
  "base64",
@@ -2696,7 +2696,7 @@ dependencies = [
 
 [[package]]
 name = "myfsio-server"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "aes-gcm",
  "async-trait",
@@ -2714,6 +2714,8 @@ dependencies = [
  "dotenvy",
  "duckdb",
  "futures",
+ "hex",
+ "http-body 1.0.1",
  "http-body-util",
  "hyper 1.9.0",
  "md-5 0.10.6",
@@ -2740,6 +2742,7 @@ dependencies = [
  "tempfile",
  "tera",
  "tokio",
+ "tokio-stream",
  "tokio-util",
  "tower",
  "tower-http",
@@ -2750,7 +2753,7 @@ dependencies = [
 
 [[package]]
 name = "myfsio-storage"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "chrono",
  "dashmap",
@@ -2773,10 +2776,11 @@ dependencies = [
 
 [[package]]
 name = "myfsio-xml"
-version = "0.4.3"
+version = "0.4.4"
 dependencies = [
  "chrono",
  "myfsio-common",
+ "percent-encoding",
  "quick-xml",
  "serde",
 ]
@@ -4193,6 +4197,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"

Cargo.toml

@@ -10,14 +10,14 @@ members = [
 ]
 
 [workspace.package]
-version = "0.4.3"
+version = "0.4.4"
 edition = "2021"
 
 [workspace.dependencies]
 tokio = { version = "1", features = ["full"] }
 axum = { version = "0.8" }
 tower = { version = "0.5" }
-tower-http = { version = "0.6", features = ["cors", "trace", "fs", "compression-gzip"] }
+tower-http = { version = "0.6", features = ["cors", "trace", "fs", "compression-gzip", "timeout"] }
 hyper = { version = "1" }
 bytes = "1"
 serde = { version = "1", features = ["derive"] }
@@ -43,6 +43,7 @@ thiserror = "2"
 chrono = { version = "0.4", features = ["serde"] }
 base64 = "0.22"
 tokio-util = { version = "0.7", features = ["io"] }
+tokio-stream = "0.1"
 futures = "0.3"
 dashmap = "6"
 crc32fast = "1"

crates/myfsio-common/src/constants.rs

@@ -8,6 +8,7 @@ pub const STATS_FILE: &str = "stats.json";
 pub const ETAG_INDEX_FILE: &str = "etag_index.json";
 pub const INDEX_FILE: &str = "_index.json";
 pub const MANIFEST_FILE: &str = "manifest.json";
+pub const DIR_MARKER_FILE: &str = ".__myfsio_dirobj__";
 
 pub const INTERNAL_FOLDERS: &[&str] = &[".meta", ".versions", ".multipart"];
 

crates/myfsio-common/src/error.rs

@@ -5,13 +5,17 @@ pub enum S3ErrorCode {
     AccessDenied,
     BadDigest,
     BucketAlreadyExists,
+    BucketAlreadyOwnedByYou,
     BucketNotEmpty,
     EntityTooLarge,
+    EntityTooSmall,
     InternalError,
     InvalidAccessKeyId,
     InvalidArgument,
     InvalidBucketName,
     InvalidKey,
+    InvalidPart,
+    InvalidPartOrder,
     InvalidPolicyDocument,
     InvalidRange,
     InvalidRequest,
@@ -19,13 +23,17 @@ pub enum S3ErrorCode {
     MalformedXML,
     MethodNotAllowed,
     NoSuchBucket,
+    NoSuchBucketPolicy,
     NoSuchKey,
+    NoSuchLifecycleConfiguration,
     NoSuchUpload,
     NoSuchVersion,
     NoSuchTagSet,
     PreconditionFailed,
     NotModified,
     QuotaExceeded,
+    RequestTimeTooSkewed,
+    ServerSideEncryptionConfigurationNotFoundError,
     SignatureDoesNotMatch,
     SlowDown,
 }
@@ -36,13 +44,17 @@ impl S3ErrorCode {
             Self::AccessDenied => 403,
             Self::BadDigest => 400,
             Self::BucketAlreadyExists => 409,
+            Self::BucketAlreadyOwnedByYou => 409,
             Self::BucketNotEmpty => 409,
             Self::EntityTooLarge => 413,
+            Self::EntityTooSmall => 400,
             Self::InternalError => 500,
             Self::InvalidAccessKeyId => 403,
             Self::InvalidArgument => 400,
             Self::InvalidBucketName => 400,
             Self::InvalidKey => 400,
+            Self::InvalidPart => 400,
+            Self::InvalidPartOrder => 400,
             Self::InvalidPolicyDocument => 400,
             Self::InvalidRange => 416,
             Self::InvalidRequest => 400,
@@ -50,15 +62,19 @@ impl S3ErrorCode {
             Self::MalformedXML => 400,
             Self::MethodNotAllowed => 405,
             Self::NoSuchBucket => 404,
+            Self::NoSuchBucketPolicy => 404,
             Self::NoSuchKey => 404,
+            Self::NoSuchLifecycleConfiguration => 404,
             Self::NoSuchUpload => 404,
             Self::NoSuchVersion => 404,
             Self::NoSuchTagSet => 404,
             Self::PreconditionFailed => 412,
             Self::NotModified => 304,
             Self::QuotaExceeded => 403,
+            Self::RequestTimeTooSkewed => 403,
+            Self::ServerSideEncryptionConfigurationNotFoundError => 404,
             Self::SignatureDoesNotMatch => 403,
-            Self::SlowDown => 429,
+            Self::SlowDown => 503,
         }
     }
 
@@ -67,13 +83,17 @@ impl S3ErrorCode {
             Self::AccessDenied => "AccessDenied",
             Self::BadDigest => "BadDigest",
             Self::BucketAlreadyExists => "BucketAlreadyExists",
+            Self::BucketAlreadyOwnedByYou => "BucketAlreadyOwnedByYou",
             Self::BucketNotEmpty => "BucketNotEmpty",
             Self::EntityTooLarge => "EntityTooLarge",
+            Self::EntityTooSmall => "EntityTooSmall",
             Self::InternalError => "InternalError",
             Self::InvalidAccessKeyId => "InvalidAccessKeyId",
             Self::InvalidArgument => "InvalidArgument",
             Self::InvalidBucketName => "InvalidBucketName",
             Self::InvalidKey => "InvalidKey",
+            Self::InvalidPart => "InvalidPart",
+            Self::InvalidPartOrder => "InvalidPartOrder",
             Self::InvalidPolicyDocument => "InvalidPolicyDocument",
             Self::InvalidRange => "InvalidRange",
             Self::InvalidRequest => "InvalidRequest",
@@ -81,13 +101,19 @@ impl S3ErrorCode {
             Self::MalformedXML => "MalformedXML",
             Self::MethodNotAllowed => "MethodNotAllowed",
             Self::NoSuchBucket => "NoSuchBucket",
+            Self::NoSuchBucketPolicy => "NoSuchBucketPolicy",
             Self::NoSuchKey => "NoSuchKey",
+            Self::NoSuchLifecycleConfiguration => "NoSuchLifecycleConfiguration",
             Self::NoSuchUpload => "NoSuchUpload",
             Self::NoSuchVersion => "NoSuchVersion",
             Self::NoSuchTagSet => "NoSuchTagSet",
             Self::PreconditionFailed => "PreconditionFailed",
             Self::NotModified => "NotModified",
             Self::QuotaExceeded => "QuotaExceeded",
+            Self::RequestTimeTooSkewed => "RequestTimeTooSkewed",
+            Self::ServerSideEncryptionConfigurationNotFoundError => {
+                "ServerSideEncryptionConfigurationNotFoundError"
+            }
             Self::SignatureDoesNotMatch => "SignatureDoesNotMatch",
             Self::SlowDown => "SlowDown",
         }
@@ -98,13 +124,17 @@ impl S3ErrorCode {
             Self::AccessDenied => "Access Denied",
             Self::BadDigest => "The Content-MD5 or checksum value you specified did not match what we received",
             Self::BucketAlreadyExists => "The requested bucket name is not available",
+            Self::BucketAlreadyOwnedByYou => "Your previous request to create the named bucket succeeded and you already own it",
             Self::BucketNotEmpty => "The bucket you tried to delete is not empty",
             Self::EntityTooLarge => "Your proposed upload exceeds the maximum allowed size",
+            Self::EntityTooSmall => "Your proposed upload is smaller than the minimum allowed object size",
             Self::InternalError => "We encountered an internal error. Please try again.",
             Self::InvalidAccessKeyId => "The access key ID you provided does not exist",
             Self::InvalidArgument => "Invalid argument",
             Self::InvalidBucketName => "The specified bucket is not valid",
             Self::InvalidKey => "The specified key is not valid",
+            Self::InvalidPart => "One or more of the specified parts could not be found",
+            Self::InvalidPartOrder => "The list of parts was not in ascending order",
             Self::InvalidPolicyDocument => "The content of the form does not meet the conditions specified in the policy document",
             Self::InvalidRange => "The requested range is not satisfiable",
             Self::InvalidRequest => "Invalid request",
@@ -112,13 +142,17 @@ impl S3ErrorCode {
             Self::MalformedXML => "The XML you provided was not well-formed",
             Self::MethodNotAllowed => "The specified method is not allowed against this resource",
             Self::NoSuchBucket => "The specified bucket does not exist",
+            Self::NoSuchBucketPolicy => "The bucket policy does not exist",
             Self::NoSuchKey => "The specified key does not exist",
+            Self::NoSuchLifecycleConfiguration => "The lifecycle configuration does not exist",
             Self::NoSuchUpload => "The specified multipart upload does not exist",
             Self::NoSuchVersion => "The specified version does not exist",
             Self::NoSuchTagSet => "The TagSet does not exist",
             Self::PreconditionFailed => "At least one of the preconditions you specified did not hold",
             Self::NotModified => "Not Modified",
             Self::QuotaExceeded => "The bucket quota has been exceeded",
+            Self::RequestTimeTooSkewed => "The difference between the request time and the server's time is too large",
+            Self::ServerSideEncryptionConfigurationNotFoundError => "The server side encryption configuration was not found",
             Self::SignatureDoesNotMatch => "The request signature we calculated does not match the signature you provided",
             Self::SlowDown => "Please reduce your request rate",
         }

crates/myfsio-common/src/types.rs

@@ -12,6 +12,10 @@ pub struct ObjectMeta {
     pub content_type: Option<String>,
     pub storage_class: Option<String>,
     pub metadata: HashMap<String, String>,
+    #[serde(default)]
+    pub version_id: Option<String>,
+    #[serde(default)]
+    pub is_delete_marker: bool,
 }
 
 impl ObjectMeta {
@@ -24,10 +28,19 @@ impl ObjectMeta {
             content_type: None,
             storage_class: Some("STANDARD".to_string()),
             metadata: HashMap::new(),
+            version_id: None,
+            is_delete_marker: false,
         }
     }
 }
 
+#[derive(Debug, Clone, Default)]
+pub struct DeleteOutcome {
+    pub version_id: Option<String>,
+    pub is_delete_marker: bool,
+    pub existed: bool,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BucketMeta {
     pub name: String,

crates/myfsio-server/Cargo.toml

@@ -23,15 +23,18 @@ serde_urlencoded = "0.7"
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 tokio-util = { workspace = true }
+tokio-stream = { workspace = true }
 chrono = { workspace = true }
 uuid = { workspace = true }
 futures = { workspace = true }
+http-body = "1"
 http-body-util = "0.1"
 percent-encoding = { workspace = true }
 quick-xml = { workspace = true }
 mime_guess = "2"
 crc32fast = { workspace = true }
 sha2 = { workspace = true }
+hex = { workspace = true }
 duckdb = { workspace = true }
 roxmltree = "0.20"
 parking_lot = { workspace = true }

crates/myfsio-server/src/config.rs

@@ -81,7 +81,12 @@ pub struct ServerConfig {
     pub multipart_min_part_size: u64,
     pub bulk_delete_max_keys: usize,
     pub stream_chunk_size: usize,
+    pub request_body_timeout_secs: u64,
     pub ratelimit_default: RateLimitSetting,
+    pub ratelimit_list_buckets: RateLimitSetting,
+    pub ratelimit_bucket_ops: RateLimitSetting,
+    pub ratelimit_object_ops: RateLimitSetting,
+    pub ratelimit_head_ops: RateLimitSetting,
     pub ratelimit_admin: RateLimitSetting,
     pub ratelimit_storage_uri: String,
     pub ui_enabled: bool,
@@ -225,8 +230,17 @@ impl ServerConfig {
         let multipart_min_part_size = parse_u64_env("MULTIPART_MIN_PART_SIZE", 5_242_880);
         let bulk_delete_max_keys = parse_usize_env("BULK_DELETE_MAX_KEYS", 1000);
         let stream_chunk_size = parse_usize_env("STREAM_CHUNK_SIZE", 1_048_576);
+        let request_body_timeout_secs = parse_u64_env("REQUEST_BODY_TIMEOUT_SECONDS", 60);
         let ratelimit_default =
-            parse_rate_limit_env("RATE_LIMIT_DEFAULT", RateLimitSetting::new(200, 60));
+            parse_rate_limit_env("RATE_LIMIT_DEFAULT", RateLimitSetting::new(500, 60));
+        let ratelimit_list_buckets =
+            parse_rate_limit_env("RATE_LIMIT_LIST_BUCKETS", ratelimit_default);
+        let ratelimit_bucket_ops =
+            parse_rate_limit_env("RATE_LIMIT_BUCKET_OPS", ratelimit_default);
+        let ratelimit_object_ops =
+            parse_rate_limit_env("RATE_LIMIT_OBJECT_OPS", ratelimit_default);
+        let ratelimit_head_ops =
+            parse_rate_limit_env("RATE_LIMIT_HEAD_OPS", ratelimit_default);
         let ratelimit_admin =
             parse_rate_limit_env("RATE_LIMIT_ADMIN", RateLimitSetting::new(60, 60));
         let ratelimit_storage_uri =
@@ -304,7 +318,12 @@ impl ServerConfig {
             multipart_min_part_size,
             bulk_delete_max_keys,
             stream_chunk_size,
+            request_body_timeout_secs,
             ratelimit_default,
+            ratelimit_list_buckets,
+            ratelimit_bucket_ops,
+            ratelimit_object_ops,
+            ratelimit_head_ops,
             ratelimit_admin,
             ratelimit_storage_uri,
             ui_enabled,
@@ -387,7 +406,12 @@ impl Default for ServerConfig {
             multipart_min_part_size: 5_242_880,
             bulk_delete_max_keys: 1000,
             stream_chunk_size: 1_048_576,
-            ratelimit_default: RateLimitSetting::new(200, 60),
+            request_body_timeout_secs: 60,
+            ratelimit_default: RateLimitSetting::new(500, 60),
+            ratelimit_list_buckets: RateLimitSetting::new(500, 60),
+            ratelimit_bucket_ops: RateLimitSetting::new(500, 60),
+            ratelimit_object_ops: RateLimitSetting::new(500, 60),
+            ratelimit_head_ops: RateLimitSetting::new(500, 60),
             ratelimit_admin: RateLimitSetting::new(60, 60),
             ratelimit_storage_uri: "memory://".to_string(),
             ui_enabled: true,
@@ -472,7 +496,31 @@ fn parse_list_env(key: &str, default: &str) -> Vec<String> {
 }
 
 pub fn parse_rate_limit(value: &str) -> Option<RateLimitSetting> {
-    let parts = value.split_whitespace().collect::<Vec<_>>();
+    let trimmed = value.trim();
+    if let Some((requests, window)) = trimmed.split_once('/') {
+        let max_requests = requests.trim().parse::<u32>().ok()?;
+        if max_requests == 0 {
+            return None;
+        }
+        let window_str = window.trim().to_ascii_lowercase();
+        let window_seconds = if let Ok(n) = window_str.parse::<u64>() {
+            if n == 0 {
+                return None;
+            }
+            n
+        } else {
+            match window_str.as_str() {
+                "s" | "sec" | "second" | "seconds" => 1,
+                "m" | "min" | "minute" | "minutes" => 60,
+                "h" | "hr" | "hour" | "hours" => 3600,
+                "d" | "day" | "days" => 86_400,
+                _ => return None,
+            }
+        };
+        return Some(RateLimitSetting::new(max_requests, window_seconds));
+    }
+
+    let parts = trimmed.split_whitespace().collect::<Vec<_>>();
     if parts.len() != 3 || !parts[1].eq_ignore_ascii_case("per") {
         return None;
     }
@@ -517,6 +565,15 @@ mod tests {
             parse_rate_limit("3 per hours"),
             Some(RateLimitSetting::new(3, 3600))
         );
+        assert_eq!(
+            parse_rate_limit("50000/60"),
+            Some(RateLimitSetting::new(50000, 60))
+        );
+        assert_eq!(
+            parse_rate_limit("100/minute"),
+            Some(RateLimitSetting::new(100, 60))
+        );
+        assert_eq!(parse_rate_limit("0/60"), None);
         assert_eq!(parse_rate_limit("0 per minute"), None);
         assert_eq!(parse_rate_limit("bad"), None);
     }
@@ -532,7 +589,7 @@ mod tests {
 
         assert_eq!(config.object_key_max_length_bytes, 1024);
         assert_eq!(config.object_tag_limit, 50);
-        assert_eq!(config.ratelimit_default, RateLimitSetting::new(200, 60));
+        assert_eq!(config.ratelimit_default, RateLimitSetting::new(500, 60));
 
         std::env::remove_var("OBJECT_TAG_LIMIT");
         std::env::remove_var("RATE_LIMIT_DEFAULT");

crates/myfsio-server/src/handlers/config.rs

@@ -218,11 +218,8 @@ pub async fn get_encryption(state: &AppState, bucket: &str) -> Response {
             } else {
                 xml_response(
                     StatusCode::NOT_FOUND,
-                    S3Error::new(
-                        S3ErrorCode::InvalidRequest,
-                        "The server side encryption configuration was not found",
-                    )
-                    .to_xml(),
+                    S3Error::from_code(S3ErrorCode::ServerSideEncryptionConfigurationNotFoundError)
+                        .to_xml(),
                 )
             }
         }
@@ -270,11 +267,7 @@ pub async fn get_lifecycle(state: &AppState, bucket: &str) -> Response {
             } else {
                 xml_response(
                     StatusCode::NOT_FOUND,
-                    S3Error::new(
-                        S3ErrorCode::NoSuchKey,
-                        "The lifecycle configuration does not exist",
-                    )
-                    .to_xml(),
+                    S3Error::from_code(S3ErrorCode::NoSuchLifecycleConfiguration).to_xml(),
                 )
             }
         }
@@ -421,7 +414,7 @@ pub async fn get_policy(state: &AppState, bucket: &str) -> Response {
             } else {
                 xml_response(
                     StatusCode::NOT_FOUND,
-                    S3Error::new(S3ErrorCode::NoSuchKey, "No bucket policy attached").to_xml(),
+                    S3Error::from_code(S3ErrorCode::NoSuchBucketPolicy).to_xml(),
                 )
             }
         }
@@ -1095,10 +1088,43 @@ pub async fn list_object_versions(
         || archived_versions.len() > archived_count;
     xml.push_str(&format!("<IsTruncated>{}</IsTruncated>", is_truncated));
 
+    let current_keys: std::collections::HashSet<String> = objects
+        .iter()
+        .take(current_count)
+        .map(|o| o.key.clone())
+        .collect();
+    let mut latest_archived_per_key: std::collections::HashMap<String, String> =
+        std::collections::HashMap::new();
+    for v in archived_versions.iter().take(archived_count) {
+        if current_keys.contains(&v.key) {
+            continue;
+        }
+        let existing = latest_archived_per_key.get(&v.key).cloned();
+        match existing {
+            None => {
+                latest_archived_per_key.insert(v.key.clone(), v.version_id.clone());
+            }
+            Some(existing_id) => {
+                let existing_ts = archived_versions
+                    .iter()
+                    .find(|x| x.key == v.key && x.version_id == existing_id)
+                    .map(|x| x.last_modified)
+                    .unwrap_or(v.last_modified);
+                if v.last_modified > existing_ts {
+                    latest_archived_per_key.insert(v.key.clone(), v.version_id.clone());
+                }
+            }
+        }
+    }
+
     for obj in objects.iter().take(current_count) {
+        let version_id = obj.version_id.clone().unwrap_or_else(|| "null".to_string());
         xml.push_str("<Version>");
         xml.push_str(&format!("<Key>{}</Key>", xml_escape(&obj.key)));
-        xml.push_str("<VersionId>null</VersionId>");
+        xml.push_str(&format!(
+            "<VersionId>{}</VersionId>",
+            xml_escape(&version_id)
+        ));
         xml.push_str("<IsLatest>true</IsLatest>");
         xml.push_str(&format!(
             "<LastModified>{}</LastModified>",
@@ -1116,23 +1142,34 @@ pub async fn list_object_versions(
     }
 
     for version in archived_versions.iter().take(archived_count) {
-        xml.push_str("<Version>");
+        let is_latest = latest_archived_per_key
+            .get(&version.key)
+            .map(|id| id == &version.version_id)
+            .unwrap_or(false);
+        let tag = if version.is_delete_marker {
+            "DeleteMarker"
+        } else {
+            "Version"
+        };
+        xml.push_str(&format!("<{}>", tag));
         xml.push_str(&format!("<Key>{}</Key>", xml_escape(&version.key)));
         xml.push_str(&format!(
             "<VersionId>{}</VersionId>",
             xml_escape(&version.version_id)
         ));
-        xml.push_str("<IsLatest>false</IsLatest>");
+        xml.push_str(&format!("<IsLatest>{}</IsLatest>", is_latest));
         xml.push_str(&format!(
             "<LastModified>{}</LastModified>",
             myfsio_xml::response::format_s3_datetime(&version.last_modified)
         ));
-        if let Some(ref etag) = version.etag {
-            xml.push_str(&format!("<ETag>\"{}\"</ETag>", xml_escape(etag)));
+        if !version.is_delete_marker {
+            if let Some(ref etag) = version.etag {
+                xml.push_str(&format!("<ETag>\"{}\"</ETag>", xml_escape(etag)));
+            }
+            xml.push_str(&format!("<Size>{}</Size>", version.size));
+            xml.push_str("<StorageClass>STANDARD</StorageClass>");
         }
-        xml.push_str(&format!("<Size>{}</Size>", version.size));
-        xml.push_str("<StorageClass>STANDARD</StorageClass>");
-        xml.push_str("</Version>");
+        xml.push_str(&format!("</{}>", tag));
     }
 
     xml.push_str("</ListVersionsResult>");
@@ -1182,6 +1219,26 @@ pub async fn put_object_tagging(state: &AppState, bucket: &str, key: &str, body:
             .to_xml(),
         );
     }
+    for tag in &tags {
+        if tag.key.is_empty() || tag.key.len() > 128 {
+            return xml_response(
+                StatusCode::BAD_REQUEST,
+                S3Error::new(S3ErrorCode::InvalidTag, "Tag key length must be 1-128").to_xml(),
+            );
+        }
+        if tag.value.len() > 256 {
+            return xml_response(
+                StatusCode::BAD_REQUEST,
+                S3Error::new(S3ErrorCode::InvalidTag, "Tag value length must be 0-256").to_xml(),
+            );
+        }
+        if tag.key.contains('=') {
+            return xml_response(
+                StatusCode::BAD_REQUEST,
+                S3Error::new(S3ErrorCode::InvalidTag, "Tag keys must not contain '='").to_xml(),
+            );
+        }
+    }
 
     match state.storage.set_object_tags(bucket, key, &tags).await {
         Ok(()) => StatusCode::OK.into_response(),

crates/myfsio-server/src/handlers/mod.rs

@@ -47,9 +47,68 @@ fn s3_error_response(err: S3Error) -> Response {
 }
 
 fn storage_err_response(err: myfsio_storage::error::StorageError) -> Response {
+    if let myfsio_storage::error::StorageError::Io(io_err) = &err {
+        if let Some(message) = crate::middleware::sha_body::sha256_mismatch_message(io_err) {
+            return bad_digest_response(message);
+        }
+        if let Some(response) = io_error_to_s3_response(io_err) {
+            return response;
+        }
+    }
+    if let myfsio_storage::error::StorageError::DeleteMarker {
+        bucket,
+        key,
+        version_id,
+    } = &err
+    {
+        let s3_err = S3Error::from_code(S3ErrorCode::NoSuchKey)
+            .with_resource(format!("/{}/{}", bucket, key))
+            .with_request_id(uuid::Uuid::new_v4().simple().to_string());
+        let status = StatusCode::from_u16(s3_err.http_status())
+            .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
+        let mut resp_headers = HeaderMap::new();
+        resp_headers.insert("x-amz-delete-marker", "true".parse().unwrap());
+        if let Ok(vid) = version_id.parse() {
+            resp_headers.insert("x-amz-version-id", vid);
+        }
+        resp_headers.insert("content-type", "application/xml".parse().unwrap());
+        return (status, resp_headers, s3_err.to_xml()).into_response();
+    }
     s3_error_response(S3Error::from(err))
 }
 
+fn io_error_to_s3_response(err: &std::io::Error) -> Option<Response> {
+    use std::io::ErrorKind;
+    let message = err.to_string();
+    let lower = message.to_ascii_lowercase();
+    let hit_collision = matches!(
+        err.kind(),
+        ErrorKind::NotADirectory
+            | ErrorKind::IsADirectory
+            | ErrorKind::AlreadyExists
+            | ErrorKind::DirectoryNotEmpty
+    ) || lower.contains("not a directory")
+        || lower.contains("is a directory")
+        || lower.contains("file exists")
+        || lower.contains("directory not empty");
+    let hit_name_too_long = matches!(err.kind(), ErrorKind::InvalidFilename)
+        || lower.contains("file name too long");
+    if !hit_collision && !hit_name_too_long {
+        return None;
+    }
+    let code = if hit_name_too_long {
+        S3ErrorCode::InvalidKey
+    } else {
+        S3ErrorCode::InvalidRequest
+    };
+    let detail = if hit_name_too_long {
+        "Object key exceeds the filesystem's per-segment length limit"
+    } else {
+        "Object key collides with an existing object path on the storage backend"
+    };
+    Some(s3_error_response(S3Error::new(code, detail)))
+}
+
 fn trigger_replication(state: &AppState, bucket: &str, key: &str, action: &str) {
     let manager = state.replication.clone();
     let bucket = bucket.to_string();
@@ -89,6 +148,7 @@ async fn ensure_object_lock_allows_write(
             Ok(())
         }
         Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => Ok(()),
+        Err(myfsio_storage::error::StorageError::DeleteMarker { .. }) => Ok(()),
         Err(err) => Err(storage_err_response(err)),
     }
 }
@@ -237,6 +297,8 @@ pub struct BucketQuery {
     pub continuation_token: Option<String>,
     #[serde(rename = "start-after")]
     pub start_after: Option<String>,
+    #[serde(rename = "encoding-type")]
+    pub encoding_type: Option<String>,
     pub uploads: Option<String>,
     pub delete: Option<String>,
     pub versioning: Option<String>,
@@ -391,6 +453,75 @@ pub async fn get_bucket(
         Some(marker.clone())
     };
 
+    if max_keys == 0 {
+        let has_any = if delimiter.is_empty() {
+            state
+                .storage
+                .list_objects(
+                    &bucket,
+                    &myfsio_common::types::ListParams {
+                        max_keys: 1,
+                        continuation_token: effective_start.clone(),
+                        prefix: if prefix.is_empty() {
+                            None
+                        } else {
+                            Some(prefix.clone())
+                        },
+                        start_after: if is_v2 {
+                            query.start_after.clone()
+                        } else {
+                            None
+                        },
+                    },
+                )
+                .await
+                .map(|r| !r.objects.is_empty())
+                .unwrap_or(false)
+        } else {
+            state
+                .storage
+                .list_objects_shallow(
+                    &bucket,
+                    &myfsio_common::types::ShallowListParams {
+                        prefix: prefix.clone(),
+                        delimiter: delimiter.clone(),
+                        max_keys: 1,
+                        continuation_token: effective_start.clone(),
+                    },
+                )
+                .await
+                .map(|r| !r.objects.is_empty() || !r.common_prefixes.is_empty())
+                .unwrap_or(false)
+        };
+        let xml = if is_v2 {
+            myfsio_xml::response::list_objects_v2_xml(
+                &bucket,
+                &prefix,
+                &delimiter,
+                0,
+                &[],
+                &[],
+                has_any,
+                query.continuation_token.as_deref(),
+                None,
+                0,
+            )
+        } else {
+            myfsio_xml::response::list_objects_v1_xml(
+                &bucket,
+                &prefix,
+                &marker,
+                &delimiter,
+                0,
+                &[],
+                &[],
+                has_any,
+                None,
+            )
+        };
+        return (StatusCode::OK, [("content-type", "application/xml")], xml).into_response();
+    }
+
     if delimiter.is_empty() {
         let params = myfsio_common::types::ListParams {
             max_keys,
@@ -408,15 +539,20 @@ pub async fn get_bucket(
         };
         match state.storage.list_objects(&bucket, &params).await {
             Ok(result) => {
-                let next_marker = result
-                    .next_continuation_token
-                    .clone()
-                    .or_else(|| result.objects.last().map(|o| o.key.clone()));
+                let next_marker = if result.is_truncated {
+                    result
+                        .next_continuation_token
+                        .clone()
+                        .or_else(|| result.objects.last().map(|o| o.key.clone()))
+                } else {
+                    None
+                };
+                let encoding_type = query.encoding_type.as_deref();
                 let xml = if is_v2 {
                     let next_token = next_marker
                         .as_deref()
                         .map(|s| URL_SAFE.encode(s.as_bytes()));
-                    myfsio_xml::response::list_objects_v2_xml(
+                    myfsio_xml::response::list_objects_v2_xml_with_encoding(
                         &bucket,
                         &prefix,
                         &delimiter,
@@ -427,9 +563,10 @@ pub async fn get_bucket(
                         query.continuation_token.as_deref(),
                         next_token.as_deref(),
                         result.objects.len(),
+                        encoding_type,
                     )
                 } else {
-                    myfsio_xml::response::list_objects_v1_xml(
+                    myfsio_xml::response::list_objects_v1_xml_with_encoding(
                         &bucket,
                         &prefix,
                         &marker,
@@ -439,6 +576,7 @@ pub async fn get_bucket(
                         &[],
                         result.is_truncated,
                         next_marker.as_deref(),
+                        encoding_type,
                     )
                 };
                 (StatusCode::OK, [("content-type", "application/xml")], xml).into_response()
@@ -454,12 +592,13 @@ pub async fn get_bucket(
         };
         match state.storage.list_objects_shallow(&bucket, &params).await {
             Ok(result) => {
+                let encoding_type = query.encoding_type.as_deref();
                 let xml = if is_v2 {
                     let next_token = result
                         .next_continuation_token
                         .as_deref()
                         .map(|s| URL_SAFE.encode(s.as_bytes()));
-                    myfsio_xml::response::list_objects_v2_xml(
+                    myfsio_xml::response::list_objects_v2_xml_with_encoding(
                         &bucket,
                         &params.prefix,
                         &delimiter,
@@ -470,9 +609,10 @@ pub async fn get_bucket(
                         query.continuation_token.as_deref(),
                         next_token.as_deref(),
                         result.objects.len() + result.common_prefixes.len(),
+                        encoding_type,
                     )
                 } else {
-                    myfsio_xml::response::list_objects_v1_xml(
+                    myfsio_xml::response::list_objects_v1_xml_with_encoding(
                         &bucket,
                         &params.prefix,
                         &marker,
@@ -482,6 +622,7 @@ pub async fn get_bucket(
                         &result.common_prefixes,
                         result.is_truncated,
                         result.next_continuation_token.as_deref(),
+                        encoding_type,
                     )
                 };
                 (StatusCode::OK, [("content-type", "application/xml")], xml).into_response()
@@ -877,6 +1018,47 @@ fn has_upload_checksum(headers: &HeaderMap) -> bool {
         || headers.contains_key("x-amz-checksum-crc32")
 }
 
+fn persist_additional_checksums(headers: &HeaderMap, metadata: &mut HashMap<String, String>) {
+    for algo in [
+        "sha256", "sha1", "crc32", "crc32c", "crc64nvme",
+    ] {
+        let header_name = format!("x-amz-checksum-{}", algo);
+        if let Some(value) = headers.get(&header_name).and_then(|v| v.to_str().ok()) {
+            let trimmed = value.trim();
+            if !trimmed.is_empty() {
+                metadata.insert(format!("__checksum_{}__", algo), trimmed.to_string());
+            }
+        }
+    }
+    if let Some(value) = headers
+        .get("x-amz-sdk-checksum-algorithm")
+        .and_then(|v| v.to_str().ok())
+    {
+        let trimmed = value.trim().to_ascii_uppercase();
+        if !trimmed.is_empty() {
+            metadata.insert("__checksum_algorithm__".to_string(), trimmed);
+        }
+    }
+}
+
+fn apply_stored_checksum_headers(resp_headers: &mut HeaderMap, metadata: &HashMap<String, String>) {
+    for algo in [
+        "sha256", "sha1", "crc32", "crc32c", "crc64nvme",
+    ] {
+        if let Some(value) = metadata.get(&format!("__checksum_{}__", algo)) {
+            if let Ok(parsed) = value.parse() {
+                resp_headers.insert(
+                    axum::http::HeaderName::from_bytes(
+                        format!("x-amz-checksum-{}", algo).as_bytes(),
+                    )
+                    .unwrap(),
+                    parsed,
+                );
+            }
+        }
+    }
+}
+
 fn validate_upload_checksums(headers: &HeaderMap, data: &[u8]) -> Result<(), Response> {
     if let Some(expected) = base64_header_bytes(headers, "content-md5")? {
         if expected.len() != 16 || Md5::digest(data).as_slice() != expected.as_slice() {
@@ -906,7 +1088,7 @@ fn validate_upload_checksums(headers: &HeaderMap, data: &[u8]) -> Result<(), Res
     Ok(())
 }
 
-async fn collect_upload_body(body: Body, aws_chunked: bool) -> Result<Vec<u8>, Response> {
+async fn collect_upload_body(body: Body, aws_chunked: bool) -> Result<bytes::Bytes, Response> {
     if aws_chunked {
         let mut reader = chunked::decode_body(body);
         let mut data = Vec::new();
@@ -916,17 +1098,21 @@ async fn collect_upload_body(body: Body, aws_chunked: bool) -> Result<Vec<u8>, R
                 "Failed to read aws-chunked request body",
             ))
         })?;
-        return Ok(data);
+        return Ok(bytes::Bytes::from(data));
     }
 
     http_body_util::BodyExt::collect(body)
         .await
-        .map(|collected| collected.to_bytes().to_vec())
-        .map_err(|_| {
-            s3_error_response(S3Error::new(
-                S3ErrorCode::InvalidRequest,
-                "Failed to read request body",
-            ))
+        .map(|collected| collected.to_bytes())
+        .map_err(|err| {
+            if let Some(message) = crate::middleware::sha_body::sha256_mismatch_message(&err) {
+                bad_digest_response(message)
+            } else {
+                s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidRequest,
+                    "Failed to read request body",
+                ))
+            }
         })
 }
 
@@ -1131,6 +1317,8 @@ pub async fn put_object(
         }
     }
 
+    persist_additional_checksums(&headers, &mut metadata);
+
     let aws_chunked = is_aws_chunked(&headers);
     let boxed: myfsio_storage::traits::AsyncReadStream = if has_upload_checksum(&headers) {
         let data = match collect_upload_body(body, aws_chunked).await {
@@ -1145,8 +1333,7 @@ pub async fn put_object(
         Box::pin(chunked::decode_body(body))
     } else {
         let stream = tokio_util::io::StreamReader::new(
-            http_body_util::BodyStream::new(body)
-                .map_ok(|frame| frame.into_data().unwrap_or_default())
+            body.into_data_stream()
                 .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)),
         );
         Box::pin(stream)
@@ -1206,10 +1393,16 @@ pub async fn put_object(
                                 resp_headers
                                     .insert("etag", format!("\"{}\"", etag).parse().unwrap());
                             }
+                            if let Some(ref vid) = meta.version_id {
+                                if let Ok(value) = vid.parse() {
+                                    resp_headers.insert("x-amz-version-id", value);
+                                }
+                            }
                             resp_headers.insert(
                                 "x-amz-server-side-encryption",
                                 enc_ctx.algorithm.as_str().parse().unwrap(),
                             );
+                            apply_stored_checksum_headers(&mut resp_headers, &enc_metadata);
                             notifications::emit_object_created(
                                 &state,
                                 &bucket,
@@ -1239,6 +1432,17 @@ pub async fn put_object(
             if let Some(ref etag) = meta.etag {
                 resp_headers.insert("etag", format!("\"{}\"", etag).parse().unwrap());
             }
+            if let Some(ref vid) = meta.version_id {
+                if let Ok(value) = vid.parse() {
+                    resp_headers.insert("x-amz-version-id", value);
+                }
+            }
+            let stored = state
+                .storage
+                .get_object_metadata(&bucket, &key)
+                .await
+                .unwrap_or_default();
+            apply_stored_checksum_headers(&mut resp_headers, &stored);
             notifications::emit_object_created(
                 &state,
                 &bucket,
@@ -1368,7 +1572,7 @@ pub async fn get_object(
             }
         };
         let file_size = file.metadata().await.map(|m| m.len()).unwrap_or(0);
-        let stream = ReaderStream::new(file);
+        let stream = ReaderStream::with_capacity(file, 256 * 1024);
         let body = Body::from_stream(stream);
 
         let meta = head_meta.clone();
@@ -1399,10 +1603,15 @@ pub async fn get_object(
             enc_info.algorithm.parse().unwrap(),
         );
         apply_stored_response_headers(&mut resp_headers, &all_meta);
+        apply_stored_checksum_headers(&mut resp_headers, &all_meta);
         if let Some(ref requested_version) = query.version_id {
             if let Ok(value) = requested_version.parse() {
                 resp_headers.insert("x-amz-version-id", value);
             }
+        } else if let Some(vid) = all_meta.get("__version_id__") {
+            if let Ok(value) = vid.parse() {
+                resp_headers.insert("x-amz-version-id", value);
+            }
         }
 
         apply_user_metadata(&mut resp_headers, &meta.metadata);
@@ -1424,7 +1633,7 @@ pub async fn get_object(
 
     match object_result {
         Ok((meta, reader)) => {
-            let stream = ReaderStream::new(reader);
+            let stream = ReaderStream::with_capacity(reader, 256 * 1024);
             let body = Body::from_stream(stream);
 
             let mut headers = HeaderMap::new();
@@ -1443,10 +1652,15 @@ pub async fn get_object(
             );
             headers.insert("accept-ranges", "bytes".parse().unwrap());
             apply_stored_response_headers(&mut headers, &all_meta);
+            apply_stored_checksum_headers(&mut headers, &all_meta);
             if let Some(ref requested_version) = query.version_id {
                 if let Ok(value) = requested_version.parse() {
                     headers.insert("x-amz-version-id", value);
                 }
+            } else if let Some(ref vid) = meta.version_id {
+                if let Ok(value) = vid.parse() {
+                    headers.insert("x-amz-version-id", value);
+                }
             }
 
             apply_user_metadata(&mut headers, &meta.metadata);
@@ -1514,10 +1728,15 @@ pub async fn delete_object(
             .delete_object_version(&bucket, &key, version_id)
             .await
         {
-            Ok(()) => {
+            Ok(outcome) => {
                 let mut resp_headers = HeaderMap::new();
-                if let Ok(value) = version_id.parse() {
-                    resp_headers.insert("x-amz-version-id", value);
+                if let Some(ref vid) = outcome.version_id {
+                    if let Ok(value) = vid.parse() {
+                        resp_headers.insert("x-amz-version-id", value);
+                    }
+                }
+                if outcome.is_delete_marker {
+                    resp_headers.insert("x-amz-delete-marker", "true".parse().unwrap());
                 }
                 notifications::emit_object_removed(&state, &bucket, &key, "", "", "", "Delete");
                 trigger_replication(&state, &bucket, &key, "delete");
@@ -1534,10 +1753,19 @@ pub async fn delete_object(
     }
 
     match state.storage.delete_object(&bucket, &key).await {
-        Ok(()) => {
+        Ok(outcome) => {
+            let mut resp_headers = HeaderMap::new();
+            if let Some(ref vid) = outcome.version_id {
+                if let Ok(value) = vid.parse() {
+                    resp_headers.insert("x-amz-version-id", value);
+                }
+            }
+            if outcome.is_delete_marker {
+                resp_headers.insert("x-amz-delete-marker", "true".parse().unwrap());
+            }
             notifications::emit_object_removed(&state, &bucket, &key, "", "", "", "Delete");
             trigger_replication(&state, &bucket, &key, "delete");
-            StatusCode::NO_CONTENT.into_response()
+            (StatusCode::NO_CONTENT, resp_headers).into_response()
         }
         Err(e) => storage_err_response(e),
     }
@@ -1596,10 +1824,15 @@ pub async fn head_object(
             );
             headers.insert("accept-ranges", "bytes".parse().unwrap());
             apply_stored_response_headers(&mut headers, &all_meta);
+            apply_stored_checksum_headers(&mut headers, &all_meta);
             if let Some(ref requested_version) = query.version_id {
                 if let Ok(value) = requested_version.parse() {
                     headers.insert("x-amz-version-id", value);
                 }
+            } else if let Some(ref vid) = meta.version_id {
+                if let Ok(value) = vid.parse() {
+                    headers.insert("x-amz-version-id", value);
+                }
             }
 
             apply_user_metadata(&mut headers, &meta.metadata);
@@ -1632,8 +1865,7 @@ async fn upload_part_handler_with_chunking(
         Box::pin(chunked::decode_body(body))
     } else {
         let stream = tokio_util::io::StreamReader::new(
-            http_body_util::BodyStream::new(body)
-                .map_ok(|frame| frame.into_data().unwrap_or_default())
+            body.into_data_stream()
                 .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)),
         );
         Box::pin(stream)
@@ -1764,6 +1996,70 @@ async fn complete_multipart_handler(
         }
     };
 
+    if parsed.parts.is_empty() {
+        return s3_error_response(S3Error::new(
+            S3ErrorCode::MalformedXML,
+            "CompleteMultipartUpload requires at least one part",
+        ));
+    }
+
+    let mut last_part_num: u32 = 0;
+    for p in &parsed.parts {
+        if p.part_number == 0 {
+            return s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidPartOrder,
+                "Part numbers must be greater than zero",
+            ));
+        }
+        if p.part_number <= last_part_num {
+            return s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidPartOrder,
+                "Parts must be specified in ascending order with no duplicates",
+            ));
+        }
+        last_part_num = p.part_number;
+    }
+
+    let stored_parts = match state.storage.list_parts(bucket, upload_id).await {
+        Ok(list) => list,
+        Err(e) => return storage_err_response(e),
+    };
+    let stored_map: HashMap<u32, (String, u64)> = stored_parts
+        .iter()
+        .map(|p| (p.part_number, (p.etag.clone(), p.size)))
+        .collect();
+    let min_part_size: u64 = state.config.multipart_min_part_size;
+    let total_parts = parsed.parts.len();
+    for (idx, p) in parsed.parts.iter().enumerate() {
+        let stored = match stored_map.get(&p.part_number) {
+            Some(s) => s,
+            None => {
+                return s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidPart,
+                    format!("Part {} not found", p.part_number),
+                ));
+            }
+        };
+        let client_etag = p.etag.trim().trim_matches('"').to_ascii_lowercase();
+        let stored_etag = stored.0.trim().trim_matches('"').to_ascii_lowercase();
+        if !client_etag.is_empty() && client_etag != stored_etag {
+            return s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidPart,
+                format!("ETag mismatch for part {}", p.part_number),
+            ));
+        }
+        let is_final = idx + 1 == total_parts;
+        if !is_final && stored.1 < min_part_size {
+            return s3_error_response(S3Error::new(
+                S3ErrorCode::EntityTooSmall,
+                format!(
+                    "Part {} is smaller than the minimum allowed size of {} bytes",
+                    p.part_number, min_part_size
+                ),
+            ));
+        }
+    }
+
     let parts: Vec<PartInfo> = parsed
         .parts
         .iter()
@@ -1852,7 +2148,8 @@ async fn object_attributes_handler(
 
     if all || attrs.contains("etag") {
         if let Some(etag) = &meta.etag {
-            xml.push_str(&format!("<ETag>{}</ETag>", xml_escape(etag)));
+            let trimmed = etag.trim_matches('"');
+            xml.push_str(&format!("<ETag>\"{}\"</ETag>", xml_escape(trimmed)));
         }
     }
     if all || attrs.contains("storageclass") {
@@ -1909,48 +2206,151 @@ async fn copy_object_handler(
         return resp;
     }
 
-    let copy_result = if let Some(version_id) = src_version_id
-        .as_deref()
-        .filter(|value| !is_null_version(Some(*value)))
-    {
-        let (_meta, mut reader) = match state
+    let metadata_directive = headers
+        .get("x-amz-metadata-directive")
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.trim().to_ascii_uppercase())
+        .unwrap_or_else(|| "COPY".to_string());
+    let tagging_directive = headers
+        .get("x-amz-tagging-directive")
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.trim().to_ascii_uppercase())
+        .unwrap_or_else(|| "COPY".to_string());
+    let replace_metadata = metadata_directive == "REPLACE";
+    let replace_tagging = tagging_directive == "REPLACE";
+
+    let same_object = src_bucket == dst_bucket
+        && src_key == dst_key
+        && src_version_id.as_deref().unwrap_or("") == "";
+    if same_object && !replace_metadata && !replace_tagging {
+        return s3_error_response(S3Error::new(
+            S3ErrorCode::InvalidRequest,
+            "This copy request is illegal because it is trying to copy an object to itself without changing the object's metadata, storage class, website redirect location or encryption attributes.",
+        ));
+    }
+
+    let source_metadata_existing = match src_version_id.as_deref() {
+        Some(version_id) if version_id != "null" => {
+            match state
+                .storage
+                .get_object_version_metadata(&src_bucket, &src_key, version_id)
+                .await
+            {
+                Ok(metadata) => metadata,
+                Err(e) => return storage_err_response(e),
+            }
+        }
+        _ => match state
             .storage
-            .get_object_version(&src_bucket, &src_key, version_id)
+            .get_object_metadata(&src_bucket, &src_key)
             .await
         {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
+    };
+
+    let dst_metadata = if replace_metadata {
+        let mut m: HashMap<String, String> = HashMap::new();
+        for (request_header, metadata_key, _) in internal_header_pairs() {
+            if let Some(value) = headers.get(*request_header).and_then(|v| v.to_str().ok()) {
+                if *request_header == "content-encoding" {
+                    if let Some(decoded_encoding) = decoded_content_encoding(value) {
+                        m.insert((*metadata_key).to_string(), decoded_encoding);
+                    }
+                } else {
+                    m.insert((*metadata_key).to_string(), value.to_string());
+                }
+            }
+        }
+        let content_type = guessed_content_type(
+            dst_key,
+            headers.get("content-type").and_then(|v| v.to_str().ok()),
+        );
+        m.insert("__content_type__".to_string(), content_type);
+        for (name, value) in headers.iter() {
+            let name_str = name.as_str();
+            if let Some(meta_key) = name_str.strip_prefix("x-amz-meta-") {
+                if let Ok(val) = value.to_str() {
+                    m.insert(meta_key.to_string(), val.to_string());
+                }
+            }
+        }
+        if let Some(value) = headers
+            .get("x-amz-storage-class")
+            .and_then(|v| v.to_str().ok())
+        {
+            m.insert("__storage_class__".to_string(), value.to_ascii_uppercase());
+        }
+        m
+    } else {
+        source_metadata_existing.clone()
+    };
+
+    let (_meta, mut reader) = match src_version_id.as_deref() {
+        Some(version_id) if version_id != "null" => {
+            match state
+                .storage
+                .get_object_version(&src_bucket, &src_key, version_id)
+                .await
+            {
+                Ok(result) => result,
+                Err(e) => return storage_err_response(e),
+            }
+        }
+        _ => match state.storage.get_object(&src_bucket, &src_key).await {
             Ok(result) => result,
             Err(e) => return storage_err_response(e),
-        };
-        let mut data = Vec::new();
-        if let Err(e) = reader.read_to_end(&mut data).await {
-            return storage_err_response(myfsio_storage::error::StorageError::Io(e));
-        }
-        let metadata = match state
-            .storage
-            .get_object_version_metadata(&src_bucket, &src_key, version_id)
-            .await
-        {
-            Ok(metadata) => metadata,
-            Err(e) => return storage_err_response(e),
-        };
-        state
-            .storage
-            .put_object(
-                dst_bucket,
-                dst_key,
-                Box::pin(std::io::Cursor::new(data)),
-                Some(metadata),
-            )
-            .await
-    } else {
-        state
-            .storage
-            .copy_object(&src_bucket, &src_key, dst_bucket, dst_key)
-            .await
+        },
     };
+    let mut data = Vec::new();
+    if let Err(e) = reader.read_to_end(&mut data).await {
+        return storage_err_response(myfsio_storage::error::StorageError::Io(e));
+    }
+
+    let copy_result = state
+        .storage
+        .put_object(
+            dst_bucket,
+            dst_key,
+            Box::pin(std::io::Cursor::new(data)),
+            Some(dst_metadata),
+        )
+        .await;
 
     match copy_result {
         Ok(meta) => {
+            if replace_tagging {
+                let tags = match headers
+                    .get("x-amz-tagging")
+                    .and_then(|value| value.to_str().ok())
+                    .map(parse_tagging_header)
+                    .transpose()
+                {
+                    Ok(tags) => tags,
+                    Err(response) => return response,
+                };
+                if let Some(ref tags) = tags {
+                    if tags.len() > state.config.object_tag_limit {
+                        return s3_error_response(S3Error::new(
+                            S3ErrorCode::InvalidTag,
+                            format!("Maximum {} tags allowed", state.config.object_tag_limit),
+                        ));
+                    }
+                    if let Err(e) = state
+                        .storage
+                        .set_object_tags(dst_bucket, dst_key, tags)
+                        .await
+                    {
+                        return storage_err_response(e);
+                    }
+                } else {
+                    let _ = state
+                        .storage
+                        .set_object_tags(dst_bucket, dst_key, &[])
+                        .await;
+                }
+            }
             let etag = meta.etag.as_deref().unwrap_or("");
             let last_modified = myfsio_xml::response::format_s3_datetime(&meta.last_modified);
             let xml = myfsio_xml::response::copy_object_result_xml(etag, &last_modified);
@@ -1983,61 +2383,117 @@ async fn delete_objects_handler(state: &AppState, bucket: &str, body: Body) -> R
         }
     };
 
-    let mut deleted = Vec::new();
-    let mut errors = Vec::new();
+    if parsed.objects.len() > 1000 {
+        return s3_error_response(S3Error::new(
+            S3ErrorCode::MalformedXML,
+            "The request must not contain more than 1000 keys",
+        ));
+    }
 
-    for obj in &parsed.objects {
-        if let Err(message) = match obj.version_id.as_deref() {
-            Some(version_id) if version_id != "null" => match state
-                .storage
-                .get_object_version_metadata(bucket, &obj.key, version_id)
-                .await
-            {
-                Ok(metadata) => object_lock::can_delete_object(&metadata, false),
-                Err(err) => Err(S3Error::from(err).message),
-            },
-            _ => match state.storage.head_object(bucket, &obj.key).await {
-                Ok(_) => match state.storage.get_object_metadata(bucket, &obj.key).await {
-                    Ok(metadata) => object_lock::can_delete_object(&metadata, false),
-                    Err(err) => Err(S3Error::from(err).message),
-                },
-                Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => Ok(()),
-                Err(err) => Err(S3Error::from(err).message),
-            },
-        } {
-            errors.push((
-                obj.key.clone(),
-                S3ErrorCode::AccessDenied.as_str().to_string(),
-                message,
-            ));
-            continue;
-        }
-        let delete_result = if let Some(version_id) = obj.version_id.as_deref() {
-            if version_id == "null" {
-                state.storage.delete_object(bucket, &obj.key).await
-            } else {
-                state
-                    .storage
-                    .delete_object_version(bucket, &obj.key, version_id)
-                    .await
-            }
-        } else {
-            state.storage.delete_object(bucket, &obj.key).await
-        };
+    use futures::stream::{self, StreamExt};
 
-        match delete_result {
-            Ok(()) => {
-                notifications::emit_object_removed(state, bucket, &obj.key, "", "", "", "Delete");
-                trigger_replication(state, bucket, &obj.key, "delete");
-                deleted.push((obj.key.clone(), obj.version_id.clone()))
+    let results: Vec<(String, Option<String>, Result<myfsio_common::types::DeleteOutcome, (String, String)>)> =
+        stream::iter(parsed.objects.iter().cloned())
+            .map(|obj| {
+                let state = state.clone();
+                let bucket = bucket.to_string();
+                async move {
+                    let key = obj.key.clone();
+                    let requested_vid = obj.version_id.clone();
+                    let lock_check: Result<(), (String, String)> = match obj.version_id.as_deref() {
+                        Some(version_id) if version_id != "null" => match state
+                            .storage
+                            .get_object_version_metadata(&bucket, &obj.key, version_id)
+                            .await
+                        {
+                            Ok(metadata) => object_lock::can_delete_object(&metadata, false)
+                                .map_err(|m| {
+                                    (S3ErrorCode::AccessDenied.as_str().to_string(), m)
+                                }),
+                            Err(err) => {
+                                let s3err = S3Error::from(err);
+                                Err((s3err.code.as_str().to_string(), s3err.message))
+                            }
+                        },
+                        _ => match state.storage.head_object(&bucket, &obj.key).await {
+                            Ok(_) => {
+                                match state
+                                    .storage
+                                    .get_object_metadata(&bucket, &obj.key)
+                                    .await
+                                {
+                                    Ok(metadata) => object_lock::can_delete_object(&metadata, false)
+                                        .map_err(|m| {
+                                            (
+                                                S3ErrorCode::AccessDenied.as_str().to_string(),
+                                                m,
+                                            )
+                                        }),
+                                    Err(err) => {
+                                        let s3err = S3Error::from(err);
+                                        Err((s3err.code.as_str().to_string(), s3err.message))
+                                    }
+                                }
+                            }
+                            Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => {
+                                Ok(())
+                            }
+                            Err(myfsio_storage::error::StorageError::DeleteMarker { .. }) => {
+                                Ok(())
+                            }
+                            Err(err) => {
+                                let s3err = S3Error::from(err);
+                                Err((s3err.code.as_str().to_string(), s3err.message))
+                            }
+                        },
+                    };
+
+                    let result = match lock_check {
+                        Err(e) => Err(e),
+                        Ok(()) => {
+                            let outcome = match obj.version_id.as_deref() {
+                                Some(version_id) if version_id != "null" => {
+                                    state
+                                        .storage
+                                        .delete_object_version(&bucket, &obj.key, version_id)
+                                        .await
+                                }
+                                _ => state.storage.delete_object(&bucket, &obj.key).await,
+                            };
+                            outcome.map_err(|e| {
+                                let s3err = S3Error::from(e);
+                                (s3err.code.as_str().to_string(), s3err.message)
+                            })
+                        }
+                    };
+                    (key, requested_vid, result)
+                }
+            })
+            .buffer_unordered(32)
+            .collect()
+            .await;
+
+    let mut deleted: Vec<myfsio_xml::response::DeletedEntry> = Vec::new();
+    let mut errors: Vec<(String, String, String)> = Vec::new();
+    for (key, requested_vid, result) in results {
+        match result {
+            Ok(outcome) => {
+                notifications::emit_object_removed(state, bucket, &key, "", "", "", "Delete");
+                trigger_replication(state, bucket, &key, "delete");
+                let delete_marker_version_id = if outcome.is_delete_marker {
+                    outcome.version_id.clone()
+                } else {
+                    None
+                };
+                deleted.push(myfsio_xml::response::DeletedEntry {
+                    key,
+                    version_id: requested_vid,
+                    delete_marker: outcome.is_delete_marker,
+                    delete_marker_version_id,
+                });
             }
-            Err(e) => {
-                let s3err = S3Error::from(e);
-                errors.push((
-                    obj.key.clone(),
-                    s3err.code.as_str().to_string(),
-                    s3err.message,
-                ));
+            Err((code, message)) => {
+                errors.push((key, code, message));
             }
         }
     }
@@ -2109,7 +2565,7 @@ async fn range_get_handler(
 
     let length = end - start + 1;
     let limited = file.take(length);
-    let stream = ReaderStream::new(limited);
+    let stream = ReaderStream::with_capacity(limited, 256 * 1024);
     let body = Body::from_stream(stream);
 
     let mut headers = HeaderMap::new();
@@ -2211,7 +2667,8 @@ async fn evaluate_put_preconditions(
             }
             None
         }
-        Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => {
+        Err(myfsio_storage::error::StorageError::ObjectNotFound { .. })
+        | Err(myfsio_storage::error::StorageError::DeleteMarker { .. }) => {
             if has_if_match {
                 Some(s3_error_response(S3Error::from_code(
                     S3ErrorCode::PreconditionFailed,
@@ -2280,9 +2737,20 @@ fn evaluate_copy_preconditions(
 }
 
 fn parse_http_date(value: &str) -> Option<DateTime<Utc>> {
-    DateTime::parse_from_rfc2822(value)
-        .ok()
-        .map(|dt| dt.with_timezone(&Utc))
+    let trimmed = value.trim();
+    if let Ok(dt) = DateTime::parse_from_rfc2822(trimmed) {
+        return Some(dt.with_timezone(&Utc));
+    }
+    if let Ok(dt) = DateTime::parse_from_rfc3339(trimmed) {
+        return Some(dt.with_timezone(&Utc));
+    }
+    if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(trimmed, "%A, %d-%b-%y %H:%M:%S GMT") {
+        return Some(naive.and_utc());
+    }
+    if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(trimmed, "%a %b %e %H:%M:%S %Y") {
+        return Some(naive.and_utc());
+    }
+    None
 }
 
 fn etag_condition_matches(condition: &str, etag: Option<&str>) -> bool {

crates/myfsio-server/src/handlers/ui.rs

@@ -117,16 +117,6 @@ pub async fn logout(Extension(session): Extension<SessionHandle>) -> Response {
     Redirect::to("/login").into_response()
 }
 
-pub async fn csrf_error_page(
-    State(state): State<AppState>,
-    Extension(session): Extension<SessionHandle>,
-) -> Response {
-    let ctx = base_context(&session, None);
-    let mut resp = render(&state, "csrf_error.html", &ctx);
-    *resp.status_mut() = StatusCode::FORBIDDEN;
-    resp
-}
-
 pub async fn root_redirect() -> Response {
     Redirect::to("/ui/buckets").into_response()
 }

crates/myfsio-server/src/handlers/ui_api.rs

@@ -121,6 +121,8 @@ fn storage_status(err: &StorageError) -> StatusCode {
         | StorageError::ObjectNotFound { .. }
         | StorageError::VersionNotFound { .. }
         | StorageError::UploadNotFound(_) => StatusCode::NOT_FOUND,
+        StorageError::DeleteMarker { .. } => StatusCode::NOT_FOUND,
+        StorageError::MethodNotAllowed(_) => StatusCode::METHOD_NOT_ALLOWED,
         StorageError::InvalidBucketName(_)
         | StorageError::InvalidObjectKey(_)
         | StorageError::InvalidRange
@@ -904,6 +906,35 @@ pub struct ListObjectsQuery {
     pub prefix: Option<String>,
     #[serde(default)]
     pub start_after: Option<String>,
+    #[serde(default)]
+    pub delimiter: Option<String>,
+}
+
+fn object_json(bucket_name: &str, o: &myfsio_common::types::ObjectMeta) -> Value {
+    json!({
+        "key": o.key,
+        "size": o.size,
+        "last_modified": o.last_modified.to_rfc3339(),
+        "last_modified_iso": o.last_modified.to_rfc3339(),
+        "last_modified_display": o.last_modified.format("%Y-%m-%d %H:%M:%S").to_string(),
+        "etag": o.etag.clone().unwrap_or_default(),
+        "storage_class": o.storage_class.clone().unwrap_or_else(|| "STANDARD".to_string()),
+        "content_type": o.content_type.clone().unwrap_or_default(),
+        "download_url": build_ui_object_url(bucket_name, &o.key, "download"),
+        "preview_url": build_ui_object_url(bucket_name, &o.key, "preview"),
+        "delete_endpoint": build_ui_object_url(bucket_name, &o.key, "delete"),
+        "presign_endpoint": build_ui_object_url(bucket_name, &o.key, "presign"),
+        "metadata_url": build_ui_object_url(bucket_name, &o.key, "metadata"),
+        "versions_endpoint": build_ui_object_url(bucket_name, &o.key, "versions"),
+        "restore_template": format!(
+            "/ui/buckets/{}/objects/{}/restore/VERSION_ID_PLACEHOLDER",
+            bucket_name,
+            encode_object_key(&o.key)
+        ),
+        "tags_url": build_ui_object_url(bucket_name, &o.key, "tags"),
+        "copy_url": build_ui_object_url(bucket_name, &o.key, "copy"),
+        "move_url": build_ui_object_url(bucket_name, &o.key, "move"),
+    })
 }
 
 pub async fn list_bucket_objects(
@@ -917,6 +948,49 @@ pub async fn list_bucket_objects(
     }
 
     let max_keys = q.max_keys.unwrap_or(1000).min(5000);
+    let versioning_enabled = state
+        .storage
+        .is_versioning_enabled(&bucket_name)
+        .await
+        .unwrap_or(false);
+    let stats = state.storage.bucket_stats(&bucket_name).await.ok();
+    let total_count = stats.as_ref().map(|s| s.objects).unwrap_or(0);
+
+    let use_shallow = q.delimiter.as_deref() == Some("/");
+
+    if use_shallow {
+        let params = myfsio_common::types::ShallowListParams {
+            prefix: q.prefix.clone().unwrap_or_default(),
+            delimiter: "/".to_string(),
+            max_keys,
+            continuation_token: q.continuation_token.clone(),
+        };
+        return match state
+            .storage
+            .list_objects_shallow(&bucket_name, &params)
+            .await
+        {
+            Ok(res) => {
+                let objects: Vec<Value> = res
+                    .objects
+                    .iter()
+                    .map(|o| object_json(&bucket_name, o))
+                    .collect();
+                Json(json!({
+                    "versioning_enabled": versioning_enabled,
+                    "total_count": total_count,
+                    "is_truncated": res.is_truncated,
+                    "next_continuation_token": res.next_continuation_token,
+                    "url_templates": url_templates_for(&bucket_name),
+                    "objects": objects,
+                    "common_prefixes": res.common_prefixes,
+                }))
+                .into_response()
+            }
+            Err(e) => storage_json_error(e),
+        };
+    }
+
     let params = ListParams {
         max_keys,
         continuation_token: q.continuation_token.clone(),
@@ -924,46 +998,12 @@ pub async fn list_bucket_objects(
         start_after: q.start_after.clone(),
     };
 
-    let versioning_enabled = state
-        .storage
-        .is_versioning_enabled(&bucket_name)
-        .await
-        .unwrap_or(false);
-
-    let stats = state.storage.bucket_stats(&bucket_name).await.ok();
-    let total_count = stats.as_ref().map(|s| s.objects).unwrap_or(0);
-
     match state.storage.list_objects(&bucket_name, &params).await {
         Ok(res) => {
             let objects: Vec<Value> = res
                 .objects
                 .iter()
-                .map(|o| {
-                    json!({
-                        "key": o.key,
-                        "size": o.size,
-                        "last_modified": o.last_modified.to_rfc3339(),
-                        "last_modified_iso": o.last_modified.to_rfc3339(),
-                        "last_modified_display": o.last_modified.format("%Y-%m-%d %H:%M:%S").to_string(),
-                        "etag": o.etag.clone().unwrap_or_default(),
-                        "storage_class": o.storage_class.clone().unwrap_or_else(|| "STANDARD".to_string()),
-                        "content_type": o.content_type.clone().unwrap_or_default(),
-                        "download_url": build_ui_object_url(&bucket_name, &o.key, "download"),
-                        "preview_url": build_ui_object_url(&bucket_name, &o.key, "preview"),
-                        "delete_endpoint": build_ui_object_url(&bucket_name, &o.key, "delete"),
-                        "presign_endpoint": build_ui_object_url(&bucket_name, &o.key, "presign"),
-                        "metadata_url": build_ui_object_url(&bucket_name, &o.key, "metadata"),
-                        "versions_endpoint": build_ui_object_url(&bucket_name, &o.key, "versions"),
-                        "restore_template": format!(
-                            "/ui/buckets/{}/objects/{}/restore/VERSION_ID_PLACEHOLDER",
-                            bucket_name,
-                            encode_object_key(&o.key)
-                        ),
-                        "tags_url": build_ui_object_url(&bucket_name, &o.key, "tags"),
-                        "copy_url": build_ui_object_url(&bucket_name, &o.key, "copy"),
-                        "move_url": build_ui_object_url(&bucket_name, &o.key, "move"),
-                    })
-                })
+                .map(|o| object_json(&bucket_name, o))
                 .collect();
 
             Json(json!({
@@ -1006,41 +1046,62 @@ pub async fn stream_bucket_objects(
     let stats = state.storage.bucket_stats(&bucket_name).await.ok();
     let total_count = stats.as_ref().map(|s| s.objects).unwrap_or(0);
 
-    let mut lines: Vec<String> = Vec::new();
-    lines.push(
-        json!({
-            "type": "meta",
-            "url_templates": url_templates_for(&bucket_name),
-            "versioning_enabled": versioning_enabled,
-        })
-        .to_string(),
-    );
-    lines.push(json!({ "type": "count", "total_count": total_count }).to_string());
-
     let use_delimiter = q.delimiter.as_deref() == Some("/");
     let prefix = q.prefix.clone().unwrap_or_default();
 
-    if use_delimiter {
-        let mut token: Option<String> = None;
-        loop {
-            let params = myfsio_common::types::ShallowListParams {
-                prefix: prefix.clone(),
-                delimiter: "/".to_string(),
-                max_keys: UI_OBJECT_BROWSER_MAX_KEYS,
-                continuation_token: token.clone(),
-            };
-            match state
-                .storage
-                .list_objects_shallow(&bucket_name, &params)
-                .await
-            {
-                Ok(res) => {
-                    for p in &res.common_prefixes {
-                        lines.push(json!({ "type": "folder", "prefix": p }).to_string());
-                    }
-                    for o in &res.objects {
-                        lines.push(
-                            json!({
+    let (tx, rx) = tokio::sync::mpsc::channel::<Result<bytes::Bytes, std::io::Error>>(64);
+
+    let meta_line = json!({
+        "type": "meta",
+        "url_templates": url_templates_for(&bucket_name),
+        "versioning_enabled": versioning_enabled,
+    })
+    .to_string()
+        + "\n";
+    let count_line = json!({ "type": "count", "total_count": total_count }).to_string() + "\n";
+
+    let storage = state.storage.clone();
+    let bucket = bucket_name.clone();
+
+    tokio::spawn(async move {
+        if tx
+            .send(Ok(bytes::Bytes::from(meta_line.into_bytes())))
+            .await
+            .is_err()
+        {
+            return;
+        }
+        if tx
+            .send(Ok(bytes::Bytes::from(count_line.into_bytes())))
+            .await
+            .is_err()
+        {
+            return;
+        }
+
+        if use_delimiter {
+            let mut token: Option<String> = None;
+            loop {
+                let params = myfsio_common::types::ShallowListParams {
+                    prefix: prefix.clone(),
+                    delimiter: "/".to_string(),
+                    max_keys: UI_OBJECT_BROWSER_MAX_KEYS,
+                    continuation_token: token.clone(),
+                };
+                match storage.list_objects_shallow(&bucket, &params).await {
+                    Ok(res) => {
+                        for p in &res.common_prefixes {
+                            let line = json!({ "type": "folder", "prefix": p }).to_string() + "\n";
+                            if tx
+                                .send(Ok(bytes::Bytes::from(line.into_bytes())))
+                                .await
+                                .is_err()
+                            {
+                                return;
+                            }
+                        }
+                        for o in &res.objects {
+                            let line = json!({
                                 "type": "object",
                                 "key": o.key,
                                 "size": o.size,
@@ -1050,38 +1111,46 @@ pub async fn stream_bucket_objects(
                                 "etag": o.etag.clone().unwrap_or_default(),
                                 "storage_class": o.storage_class.clone().unwrap_or_else(|| "STANDARD".to_string()),
                             })
-                            .to_string(),
-                        );
+                            .to_string()
+                                + "\n";
+                            if tx
+                                .send(Ok(bytes::Bytes::from(line.into_bytes())))
+                                .await
+                                .is_err()
+                            {
+                                return;
+                            }
+                        }
+                        if !res.is_truncated || res.next_continuation_token.is_none() {
+                            break;
+                        }
+                        token = res.next_continuation_token;
                     }
-                    if !res.is_truncated || res.next_continuation_token.is_none() {
-                        break;
+                    Err(e) => {
+                        let line =
+                            json!({ "type": "error", "error": e.to_string() }).to_string() + "\n";
+                        let _ = tx.send(Ok(bytes::Bytes::from(line.into_bytes()))).await;
+                        return;
                     }
-                    token = res.next_continuation_token;
-                }
-                Err(e) => {
-                    lines.push(json!({ "type": "error", "error": e.to_string() }).to_string());
-                    break;
                 }
             }
-        }
-    } else {
-        let mut token: Option<String> = None;
-        loop {
-            let params = ListParams {
-                max_keys: 1000,
-                continuation_token: token.clone(),
-                prefix: if prefix.is_empty() {
-                    None
-                } else {
-                    Some(prefix.clone())
-                },
-                start_after: None,
-            };
-            match state.storage.list_objects(&bucket_name, &params).await {
-                Ok(res) => {
-                    for o in &res.objects {
-                        lines.push(
-                            json!({
+        } else {
+            let mut token: Option<String> = None;
+            loop {
+                let params = ListParams {
+                    max_keys: 1000,
+                    continuation_token: token.clone(),
+                    prefix: if prefix.is_empty() {
+                        None
+                    } else {
+                        Some(prefix.clone())
+                    },
+                    start_after: None,
+                };
+                match storage.list_objects(&bucket, &params).await {
+                    Ok(res) => {
+                        for o in &res.objects {
+                            let line = json!({
                                 "type": "object",
                                 "key": o.key,
                                 "size": o.size,
@@ -1091,30 +1160,48 @@ pub async fn stream_bucket_objects(
                                 "etag": o.etag.clone().unwrap_or_default(),
                                 "storage_class": o.storage_class.clone().unwrap_or_else(|| "STANDARD".to_string()),
                             })
-                            .to_string(),
-                        );
+                            .to_string()
+                                + "\n";
+                            if tx
+                                .send(Ok(bytes::Bytes::from(line.into_bytes())))
+                                .await
+                                .is_err()
+                            {
+                                return;
+                            }
+                        }
+                        if !res.is_truncated || res.next_continuation_token.is_none() {
+                            break;
+                        }
+                        token = res.next_continuation_token;
                     }
-                    if !res.is_truncated || res.next_continuation_token.is_none() {
-                        break;
+                    Err(e) => {
+                        let line =
+                            json!({ "type": "error", "error": e.to_string() }).to_string() + "\n";
+                        let _ = tx.send(Ok(bytes::Bytes::from(line.into_bytes()))).await;
+                        return;
                     }
-                    token = res.next_continuation_token;
-                }
-                Err(e) => {
-                    lines.push(json!({ "type": "error", "error": e.to_string() }).to_string());
-                    break;
                 }
             }
         }
-    }
 
-    lines.push(json!({ "type": "done" }).to_string());
+        let done_line = json!({ "type": "done" }).to_string() + "\n";
+        let _ = tx
+            .send(Ok(bytes::Bytes::from(done_line.into_bytes())))
+            .await;
+    });
+
+    let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+    let body = Body::from_stream(stream);
 
-    let body = lines.join("\n") + "\n";
     let mut headers = HeaderMap::new();
     headers.insert(
         header::CONTENT_TYPE,
         "application/x-ndjson; charset=utf-8".parse().unwrap(),
     );
+    headers.insert(header::CACHE_CONTROL, "no-cache".parse().unwrap());
+    headers.insert("x-accel-buffering", "no".parse().unwrap());
+
     (StatusCode::OK, headers, body).into_response()
 }
 
@@ -2514,7 +2601,7 @@ async fn move_object_json(state: &AppState, bucket: &str, key: &str, body: Body)
 
     match state.storage.copy_object(bucket, key, dest_bucket, dest_key).await {
         Ok(_) => match state.storage.delete_object(bucket, key).await {
-            Ok(()) => {
+            Ok(_) => {
                 super::trigger_replication(state, dest_bucket, dest_key, "write");
                 super::trigger_replication(state, bucket, key, "delete");
                 Json(json!({
@@ -2589,7 +2676,7 @@ async fn delete_object_json(
     }
 
     match state.storage.delete_object(bucket, key).await {
-        Ok(()) => {
+        Ok(_) => {
             super::trigger_replication(state, bucket, key, "delete");
             Json(json!({
                 "status": "ok",
@@ -2868,7 +2955,7 @@ pub async fn bulk_delete_objects(
 
     for key in keys {
         match state.storage.delete_object(&bucket_name, &key).await {
-            Ok(()) => {
+            Ok(_) => {
                 super::trigger_replication(&state, &bucket_name, &key, "delete");
                 if payload.purge_versions {
                     if let Err(err) =

crates/myfsio-server/src/handlers/ui_pages.rs

@@ -227,9 +227,7 @@ async fn parse_form_any(
     if is_multipart {
         let boundary = multer::parse_boundary(&content_type)
             .map_err(|_| "Missing multipart boundary".to_string())?;
-        let stream = futures::stream::once(async move {
-            Ok::<_, std::io::Error>(bytes)
-        });
+        let stream = futures::stream::once(async move { Ok::<_, std::io::Error>(bytes) });
         let mut multipart = multer::Multipart::new(stream, boundary);
         let mut out = HashMap::new();
         while let Some(field) = multipart
@@ -2173,10 +2171,7 @@ pub async fn create_bucket(
     let wants_json = wants_json(&headers);
     let form = match parse_form_any(&headers, body).await {
         Ok(fields) => CreateBucketForm {
-            bucket_name: fields
-                .get("bucket_name")
-                .cloned()
-                .unwrap_or_default(),
+            bucket_name: fields.get("bucket_name").cloned().unwrap_or_default(),
             csrf_token: fields.get("csrf_token").cloned().unwrap_or_default(),
         },
         Err(message) => {

crates/myfsio-server/src/lib.rs

@@ -304,8 +304,7 @@ pub fn create_ui_router(state: state::AppState) -> Router {
 
     let public = Router::new()
         .route("/login", get(ui::login_page).post(ui::login_submit))
-        .route("/logout", post(ui::logout).get(ui::logout))
-        .route("/csrf-error", get(ui::csrf_error_page));
+        .route("/logout", post(ui::logout).get(ui::logout));
 
     let session_state = middleware::SessionLayerState {
         store: state.sessions.clone(),
@@ -317,7 +316,10 @@ pub fn create_ui_router(state: state::AppState) -> Router {
     protected
         .merge(public)
         .fallback(ui::not_found_page)
-        .layer(axum::middleware::from_fn(middleware::csrf_layer))
+        .layer(axum::middleware::from_fn_with_state(
+            state.clone(),
+            middleware::csrf_layer,
+        ))
         .layer(axum::middleware::from_fn_with_state(
             session_state,
             middleware::session_layer,
@@ -333,8 +335,12 @@ pub fn create_ui_router(state: state::AppState) -> Router {
 }
 
 pub fn create_router(state: state::AppState) -> Router {
-    let default_rate_limit = middleware::RateLimitLayerState::new(
+    let default_rate_limit = middleware::RateLimitLayerState::with_per_op(
         state.config.ratelimit_default,
+        state.config.ratelimit_list_buckets,
+        state.config.ratelimit_bucket_ops,
+        state.config.ratelimit_object_ops,
+        state.config.ratelimit_head_ops,
         state.config.num_trusted_proxies,
     );
     let admin_rate_limit = middleware::RateLimitLayerState::new(
@@ -575,11 +581,17 @@ pub fn create_router(state: state::AppState) -> Router {
             middleware::rate_limit_layer,
         ));
 
+    let request_body_timeout =
+        std::time::Duration::from_secs(state.config.request_body_timeout_secs);
+
     api_router
         .merge(admin_router)
         .layer(axum::middleware::from_fn(middleware::server_header))
         .layer(cors_layer(&state.config))
         .layer(tower_http::compression::CompressionLayer::new())
+        .layer(tower_http::timeout::RequestBodyTimeoutLayer::new(
+            request_body_timeout,
+        ))
         .with_state(state)
 }
 

crates/myfsio-server/src/main.rs

@@ -189,6 +189,11 @@ async fn main() {
 
     let shutdown = shutdown_signal_shared();
     let api_shutdown = shutdown.clone();
+    let api_listener = axum::serve::ListenerExt::tap_io(api_listener, |stream| {
+        if let Err(err) = stream.set_nodelay(true) {
+            tracing::trace!("failed to set TCP_NODELAY on api socket: {}", err);
+        }
+    });
     let api_task = tokio::spawn(async move {
         axum::serve(
             api_listener,
@@ -202,6 +207,11 @@ async fn main() {
 
     let ui_task = if let (Some(listener), Some(app)) = (ui_listener, ui_app) {
         let ui_shutdown = shutdown.clone();
+        let listener = axum::serve::ListenerExt::tap_io(listener, |stream| {
+            if let Err(err) = stream.set_nodelay(true) {
+                tracing::trace!("failed to set TCP_NODELAY on ui socket: {}", err);
+            }
+        });
         Some(tokio::spawn(async move {
             axum::serve(listener, app)
                 .with_graceful_shutdown(async move {

crates/myfsio-server/src/middleware/auth.rs

@@ -12,9 +12,36 @@ use serde_json::Value;
 use std::time::Instant;
 use tokio::io::AsyncReadExt;
 
+use crate::middleware::sha_body::{is_hex_sha256, Sha256VerifyBody};
 use crate::services::acl::acl_from_bucket_config;
 use crate::state::AppState;
 
+fn wrap_body_for_sha256_verification(req: &mut Request) {
+    let declared = match req
+        .headers()
+        .get("x-amz-content-sha256")
+        .and_then(|v| v.to_str().ok())
+    {
+        Some(v) => v.to_string(),
+        None => return,
+    };
+    if !is_hex_sha256(&declared) {
+        return;
+    }
+    let is_chunked = req
+        .headers()
+        .get("content-encoding")
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.to_ascii_lowercase().contains("aws-chunked"))
+        .unwrap_or(false);
+    if is_chunked {
+        return;
+    }
+    let body = std::mem::replace(req.body_mut(), axum::body::Body::empty());
+    let wrapped = Sha256VerifyBody::new(body, declared);
+    *req.body_mut() = axum::body::Body::new(wrapped);
+}
+
 #[derive(Clone, Debug)]
 struct OriginalCanonicalPath(String);
 
@@ -475,6 +502,7 @@ pub async fn auth_layer(State(state): State<AppState>, mut req: Request, next: N
                     error_response(err, &auth_path)
                 } else {
                     req.extensions_mut().insert(principal);
+                    wrap_body_for_sha256_verification(&mut req);
                     next.run(req).await
                 }
             }
@@ -1102,7 +1130,9 @@ fn verify_sigv4_header(state: &AppState, req: &Request, auth_str: &str) -> AuthR
     let parts: Vec<&str> = auth_str
         .strip_prefix("AWS4-HMAC-SHA256 ")
         .unwrap()
-        .split(", ")
+        .split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
         .collect();
 
     if parts.len() != 3 {
@@ -1112,9 +1142,24 @@ fn verify_sigv4_header(state: &AppState, req: &Request, auth_str: &str) -> AuthR
         ));
     }
 
-    let credential = parts[0].strip_prefix("Credential=").unwrap_or("");
-    let signed_headers_str = parts[1].strip_prefix("SignedHeaders=").unwrap_or("");
-    let provided_signature = parts[2].strip_prefix("Signature=").unwrap_or("");
+    let mut credential: &str = "";
+    let mut signed_headers_str: &str = "";
+    let mut provided_signature: &str = "";
+    for part in &parts {
+        if let Some(v) = part.strip_prefix("Credential=") {
+            credential = v;
+        } else if let Some(v) = part.strip_prefix("SignedHeaders=") {
+            signed_headers_str = v;
+        } else if let Some(v) = part.strip_prefix("Signature=") {
+            provided_signature = v;
+        }
+    }
+    if credential.is_empty() || signed_headers_str.is_empty() || provided_signature.is_empty() {
+        return AuthResult::Denied(S3Error::new(
+            S3ErrorCode::InvalidArgument,
+            "Malformed Authorization header",
+        ));
+    }
 
     let cred_parts: Vec<&str> = credential.split('/').collect();
     if cred_parts.len() != 5 {
@@ -1299,7 +1344,7 @@ fn verify_sigv4_query(state: &AppState, req: &Request) -> AuthResult {
         }
         if elapsed < -(state.config.sigv4_timestamp_tolerance_secs as i64) {
             return AuthResult::Denied(S3Error::new(
-                S3ErrorCode::AccessDenied,
+                S3ErrorCode::RequestTimeTooSkewed,
                 "Request is too far in the future",
             ));
         }
@@ -1369,8 +1414,11 @@ fn check_timestamp_freshness(amz_date: &str, tolerance_secs: u64) -> Option<S3Er
 
     if diff > tolerance_secs {
         return Some(S3Error::new(
-            S3ErrorCode::AccessDenied,
-            "Request timestamp too old or too far in the future",
+            S3ErrorCode::RequestTimeTooSkewed,
+            format!(
+                "The difference between the request time and the server's time is too large ({}s, tolerance {}s)",
+                diff, tolerance_secs
+            ),
         ));
     }
     None

crates/myfsio-server/src/middleware/mod.rs

@@ -1,6 +1,7 @@
 mod auth;
 pub mod ratelimit;
 pub mod session;
+pub(crate) mod sha_body;
 
 pub use auth::auth_layer;
 pub use ratelimit::{rate_limit_layer, RateLimitLayerState};

crates/myfsio-server/src/middleware/ratelimit.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use axum::extract::{ConnectInfo, Request, State};
-use axum::http::{header, StatusCode};
+use axum::http::{header, Method, StatusCode};
 use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 use parking_lot::Mutex;
@@ -13,17 +13,77 @@ use crate::config::RateLimitSetting;
 
 #[derive(Clone)]
 pub struct RateLimitLayerState {
-    limiter: Arc<FixedWindowLimiter>,
+    default_limiter: Arc<FixedWindowLimiter>,
+    list_buckets_limiter: Option<Arc<FixedWindowLimiter>>,
+    bucket_ops_limiter: Option<Arc<FixedWindowLimiter>>,
+    object_ops_limiter: Option<Arc<FixedWindowLimiter>>,
+    head_ops_limiter: Option<Arc<FixedWindowLimiter>>,
     num_trusted_proxies: usize,
 }
 
 impl RateLimitLayerState {
     pub fn new(setting: RateLimitSetting, num_trusted_proxies: usize) -> Self {
         Self {
-            limiter: Arc::new(FixedWindowLimiter::new(setting)),
+            default_limiter: Arc::new(FixedWindowLimiter::new(setting)),
+            list_buckets_limiter: None,
+            bucket_ops_limiter: None,
+            object_ops_limiter: None,
+            head_ops_limiter: None,
             num_trusted_proxies,
         }
     }
+
+    pub fn with_per_op(
+        default: RateLimitSetting,
+        list_buckets: RateLimitSetting,
+        bucket_ops: RateLimitSetting,
+        object_ops: RateLimitSetting,
+        head_ops: RateLimitSetting,
+        num_trusted_proxies: usize,
+    ) -> Self {
+        Self {
+            default_limiter: Arc::new(FixedWindowLimiter::new(default)),
+            list_buckets_limiter: (list_buckets != default)
+                .then(|| Arc::new(FixedWindowLimiter::new(list_buckets))),
+            bucket_ops_limiter: (bucket_ops != default)
+                .then(|| Arc::new(FixedWindowLimiter::new(bucket_ops))),
+            object_ops_limiter: (object_ops != default)
+                .then(|| Arc::new(FixedWindowLimiter::new(object_ops))),
+            head_ops_limiter: (head_ops != default)
+                .then(|| Arc::new(FixedWindowLimiter::new(head_ops))),
+            num_trusted_proxies,
+        }
+    }
+
+    fn select_limiter(&self, req: &Request) -> &Arc<FixedWindowLimiter> {
+        let path = req.uri().path();
+        let method = req.method();
+        if path == "/" && *method == Method::GET {
+            if let Some(ref limiter) = self.list_buckets_limiter {
+                return limiter;
+            }
+        }
+        let segments: Vec<&str> = path
+            .trim_start_matches('/')
+            .split('/')
+            .filter(|s| !s.is_empty())
+            .collect();
+        if *method == Method::HEAD {
+            if let Some(ref limiter) = self.head_ops_limiter {
+                return limiter;
+            }
+        }
+        if segments.len() == 1 {
+            if let Some(ref limiter) = self.bucket_ops_limiter {
+                return limiter;
+            }
+        } else if segments.len() >= 2 {
+            if let Some(ref limiter) = self.object_ops_limiter {
+                return limiter;
+            }
+        }
+        &self.default_limiter
+    }
 }
 
 #[derive(Debug)]
@@ -99,22 +159,34 @@ pub async fn rate_limit_layer(
     next: Next,
 ) -> Response {
     let key = rate_limit_key(&req, state.num_trusted_proxies);
-    match state.limiter.check(&key) {
+    let limiter = state.select_limiter(&req);
+    match limiter.check(&key) {
         Ok(()) => next.run(req).await,
-        Err(retry_after) => too_many_requests(retry_after),
+        Err(retry_after) => {
+            let resource = req.uri().path().to_string();
+            too_many_requests(retry_after, &resource)
+        }
     }
 }
 
-fn too_many_requests(retry_after: u64) -> Response {
-    (
-        StatusCode::TOO_MANY_REQUESTS,
+fn too_many_requests(retry_after: u64, resource: &str) -> Response {
+    let request_id = uuid::Uuid::new_v4().simple().to_string();
+    let body = myfsio_xml::response::rate_limit_exceeded_xml(resource, &request_id);
+    let mut response = (
+        StatusCode::SERVICE_UNAVAILABLE,
         [
             (header::CONTENT_TYPE, "application/xml".to_string()),
             (header::RETRY_AFTER, retry_after.to_string()),
         ],
-        myfsio_xml::response::rate_limit_exceeded_xml(),
+        body,
     )
-        .into_response()
+        .into_response();
+    if let Ok(value) = request_id.parse() {
+        response
+            .headers_mut()
+            .insert("x-amz-request-id", value);
+    }
+    response
 }
 
 fn rate_limit_key(req: &Request, num_trusted_proxies: usize) -> String {

crates/myfsio-server/src/middleware/session.rs

@@ -90,7 +90,11 @@ pub async fn session_layer(
     resp
 }
 
-pub async fn csrf_layer(req: Request, next: Next) -> Response {
+pub async fn csrf_layer(
+    State(state): State<crate::state::AppState>,
+    req: Request,
+    next: Next,
+) -> Response {
     const CSRF_HEADER_ALIAS: &str = "x-csrftoken";
 
     let method = req.method().clone();
@@ -169,7 +173,32 @@ pub async fn csrf_layer(req: Request, next: Next) -> Response {
         header_present = header_token.is_some(),
         "CSRF token mismatch"
     );
-    (StatusCode::FORBIDDEN, "Invalid CSRF token").into_response()
+
+    let accept = parts
+        .headers
+        .get(header::ACCEPT)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("");
+    let is_form_submit = content_type.starts_with("application/x-www-form-urlencoded")
+        || content_type.starts_with("multipart/form-data");
+    let wants_json =
+        accept.contains("application/json") || content_type.starts_with("application/json");
+
+    if is_form_submit && !wants_json {
+        let ctx = crate::handlers::ui::base_context(&handle, None);
+        let mut resp = crate::handlers::ui::render(&state, "csrf_error.html", &ctx);
+        *resp.status_mut() = StatusCode::FORBIDDEN;
+        return resp;
+    }
+
+    let mut resp = (
+        StatusCode::FORBIDDEN,
+        [(header::CONTENT_TYPE, "application/json")],
+        r#"{"error":"Invalid CSRF token"}"#,
+    )
+        .into_response();
+    *resp.status_mut() = StatusCode::FORBIDDEN;
+    resp
 }
 
 fn extract_multipart_token(content_type: &str, body: &[u8]) -> Option<String> {

crates/myfsio-server/src/middleware/sha_body.rs

@@ -0,0 +1,107 @@
+use axum::body::Body;
+use bytes::Bytes;
+use http_body::{Body as HttpBody, Frame};
+use sha2::{Digest, Sha256};
+use std::error::Error;
+use std::fmt;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+struct Sha256MismatchError {
+    expected: String,
+    computed: String,
+}
+
+impl Sha256MismatchError {
+    fn message(&self) -> String {
+        format!(
+            "The x-amz-content-sha256 you specified did not match what we received (expected {}, computed {})",
+            self.expected, self.computed
+        )
+    }
+}
+
+impl fmt::Display for Sha256MismatchError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "XAmzContentSHA256Mismatch: expected {}, computed {}",
+            self.expected, self.computed
+        )
+    }
+}
+
+impl Error for Sha256MismatchError {}
+
+pub struct Sha256VerifyBody {
+    inner: Body,
+    expected: String,
+    hasher: Option<Sha256>,
+}
+
+impl Sha256VerifyBody {
+    pub fn new(inner: Body, expected_hex: String) -> Self {
+        Self {
+            inner,
+            expected: expected_hex.to_ascii_lowercase(),
+            hasher: Some(Sha256::new()),
+        }
+    }
+}
+
+impl HttpBody for Sha256VerifyBody {
+    type Data = Bytes;
+    type Error = Box<dyn std::error::Error + Send + Sync>;
+
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        let this = self.as_mut().get_mut();
+        match Pin::new(&mut this.inner).poll_frame(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(Box::new(e)))),
+            Poll::Ready(Some(Ok(frame))) => {
+                if let Some(data) = frame.data_ref() {
+                    if let Some(h) = this.hasher.as_mut() {
+                        h.update(data);
+                    }
+                }
+                Poll::Ready(Some(Ok(frame)))
+            }
+            Poll::Ready(None) => {
+                if let Some(hasher) = this.hasher.take() {
+                    let computed = hex::encode(hasher.finalize());
+                    if computed != this.expected {
+                        return Poll::Ready(Some(Err(Box::new(Sha256MismatchError {
+                            expected: this.expected.clone(),
+                            computed,
+                        }))));
+                    }
+                }
+                Poll::Ready(None)
+            }
+        }
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> http_body::SizeHint {
+        self.inner.size_hint()
+    }
+}
+
+pub fn is_hex_sha256(s: &str) -> bool {
+    s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit())
+}
+
+pub fn sha256_mismatch_message(err: &(dyn Error + 'static)) -> Option<String> {
+    if let Some(mismatch) = err.downcast_ref::<Sha256MismatchError>() {
+        return Some(mismatch.message());
+    }
+
+    err.source().and_then(sha256_mismatch_message)
+}

crates/myfsio-server/tests/integration.rs

@@ -1,5 +1,7 @@
 use axum::body::Body;
 use axum::http::{Method, Request, StatusCode};
+use base64::engine::general_purpose::URL_SAFE;
+use base64::Engine;
 use http_body_util::BodyExt;
 use myfsio_storage::traits::{AsyncReadStream, StorageEngine};
 use serde_json::Value;
@@ -53,6 +55,7 @@ fn test_app_with_iam(iam_json: serde_json::Value) -> (axum::Router, tempfile::Te
         ui_enabled: false,
         templates_dir: std::path::PathBuf::from("templates"),
         static_dir: std::path::PathBuf::from("static"),
+        multipart_min_part_size: 1,
         ..myfsio_server::config::ServerConfig::default()
     };
     let state = myfsio_server::state::AppState::new(config);
@@ -118,6 +121,10 @@ fn test_app_with_rate_limits(
         storage_root: tmp.path().to_path_buf(),
         iam_config_path: iam_path.join("iam.json"),
         ratelimit_default: default,
+        ratelimit_list_buckets: default,
+        ratelimit_bucket_ops: default,
+        ratelimit_object_ops: default,
+        ratelimit_head_ops: default,
         ratelimit_admin: admin,
         ui_enabled: false,
         ..myfsio_server::config::ServerConfig::default()
@@ -156,7 +163,7 @@ async fn rate_limit_default_and_admin_are_independent() {
         )
         .await
         .unwrap();
-    assert_eq!(second.status(), StatusCode::TOO_MANY_REQUESTS);
+    assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE);
     assert!(second.headers().contains_key("retry-after"));
 
     let admin_first = app
@@ -192,7 +199,7 @@ async fn rate_limit_default_and_admin_are_independent() {
         )
         .await
         .unwrap();
-    assert_eq!(admin_third.status(), StatusCode::TOO_MANY_REQUESTS);
+    assert_eq!(admin_third.status(), StatusCode::SERVICE_UNAVAILABLE);
 }
 
 fn test_ui_state() -> (myfsio_server::state::AppState, tempfile::TempDir) {
@@ -2304,9 +2311,16 @@ async fn test_versioned_object_can_be_read_and_deleted_by_version_id() {
     )
     .unwrap();
     let archived_version_id = list_body
-        .split("<VersionId>")
-        .filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
-        .find(|id| *id != "null")
+        .split("<Version>")
+        .skip(1)
+        .find(|block| block.contains("<IsLatest>false</IsLatest>"))
+        .and_then(|block| {
+            block
+                .split("<VersionId>")
+                .nth(1)
+                .and_then(|s| s.split_once("</VersionId>").map(|(id, _)| id))
+        })
+        .filter(|id| *id != "null")
         .expect("archived version id")
         .to_string();
 
@@ -2394,6 +2408,457 @@ async fn test_versioned_object_can_be_read_and_deleted_by_version_id() {
     assert_eq!(missing_resp.status(), StatusCode::NOT_FOUND);
 }
 
+#[tokio::test]
+async fn test_versioned_put_and_delete_emit_version_headers_and_delete_markers() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/compat-bucket", Body::empty()))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/compat-bucket?versioning")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::from(
+                    "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    let put_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/compat-bucket/doc.txt",
+            Body::from("first"),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(put_resp.status(), StatusCode::OK);
+    let first_version = put_resp
+        .headers()
+        .get("x-amz-version-id")
+        .expect("PUT on versioned bucket must emit x-amz-version-id")
+        .to_str()
+        .unwrap()
+        .to_string();
+    assert!(!first_version.is_empty());
+
+    let overwrite_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/compat-bucket/doc.txt",
+            Body::from("second"),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(overwrite_resp.status(), StatusCode::OK);
+    let second_version = overwrite_resp
+        .headers()
+        .get("x-amz-version-id")
+        .expect("overwrite on versioned bucket must emit a new x-amz-version-id")
+        .to_str()
+        .unwrap()
+        .to_string();
+    assert_ne!(first_version, second_version);
+
+    let delete_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            "/compat-bucket/doc.txt",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(delete_resp.status(), StatusCode::NO_CONTENT);
+    assert_eq!(
+        delete_resp
+            .headers()
+            .get("x-amz-delete-marker")
+            .and_then(|v| v.to_str().ok()),
+        Some("true")
+    );
+    assert!(delete_resp.headers().contains_key("x-amz-version-id"));
+
+    let versions_resp = app
+        .oneshot(signed_request(
+            Method::GET,
+            "/compat-bucket?versions",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let versions_body = String::from_utf8(
+        versions_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(
+        versions_body.contains("<DeleteMarker>"),
+        "expected DeleteMarker entry in ListObjectVersions output, got: {}",
+        versions_body
+    );
+}
+
+#[tokio::test]
+async fn test_consecutive_slashes_in_key_round_trip() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/slashes-bucket", Body::empty()))
+        .await
+        .unwrap();
+
+    let put_ab = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/slashes-bucket/a/b",
+            Body::from("single"),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(put_ab.status(), StatusCode::OK);
+
+    let put_double = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/slashes-bucket/a//b",
+            Body::from("double"),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(put_double.status(), StatusCode::OK);
+
+    let put_triple = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/slashes-bucket/a///b",
+            Body::from("triple"),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(put_triple.status(), StatusCode::OK);
+
+    let get_ab = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/slashes-bucket/a/b",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(get_ab.status(), StatusCode::OK);
+    let body_ab = get_ab.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body_ab[..], b"single");
+
+    let get_triple = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/slashes-bucket/a///b",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(get_triple.status(), StatusCode::OK);
+    let body_triple = get_triple.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body_triple[..], b"triple");
+
+    let list_resp = app
+        .oneshot(signed_request(
+            Method::GET,
+            "/slashes-bucket?list-type=2",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let list_body = String::from_utf8(
+        list_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(
+        list_body.contains("<Key>a/b</Key>"),
+        "expected a/b in listing: {}",
+        list_body
+    );
+    assert!(
+        list_body.contains("<Key>a//b</Key>"),
+        "expected a//b in listing: {}",
+        list_body
+    );
+    assert!(
+        list_body.contains("<Key>a///b</Key>"),
+        "expected a///b in listing: {}",
+        list_body
+    );
+}
+
+#[tokio::test]
+async fn test_delete_live_version_restores_previous_to_live_slot() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/restore-bucket", Body::empty()))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/restore-bucket?versioning")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::from(
+                    "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    let v1_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/restore-bucket/k",
+            Body::from("one"),
+        ))
+        .await
+        .unwrap();
+    let v1 = v1_resp
+        .headers()
+        .get("x-amz-version-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_string();
+
+    let v2_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/restore-bucket/k",
+            Body::from("two"),
+        ))
+        .await
+        .unwrap();
+    let v2 = v2_resp
+        .headers()
+        .get("x-amz-version-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_string();
+    assert_ne!(v1, v2);
+
+    let del = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            &format!("/restore-bucket/k?versionId={}", v2),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(del.status(), StatusCode::NO_CONTENT);
+
+    let get_live = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/restore-bucket/k",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(get_live.status(), StatusCode::OK);
+    let body = get_live.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body[..], b"one");
+
+    let get_v1 = app
+        .oneshot(signed_request(
+            Method::GET,
+            &format!("/restore-bucket/k?versionId={}", v1),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(get_v1.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_delete_active_delete_marker_restores_previous_to_live_slot() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/undel-bucket", Body::empty()))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/undel-bucket?versioning")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::from(
+                    "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/undel-bucket/k",
+            Body::from("only"),
+        ))
+        .await
+        .unwrap();
+
+    let del = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            "/undel-bucket/k",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let dm_version = del
+        .headers()
+        .get("x-amz-version-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_string();
+    assert_eq!(
+        del.headers()
+            .get("x-amz-delete-marker")
+            .and_then(|v| v.to_str().ok()),
+        Some("true")
+    );
+
+    let shadowed = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/undel-bucket/k",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(shadowed.status(), StatusCode::NOT_FOUND);
+
+    let del_dm = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            &format!("/undel-bucket/k?versionId={}", dm_version),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(del_dm.status(), StatusCode::NO_CONTENT);
+
+    let restored = app
+        .oneshot(signed_request(
+            Method::GET,
+            "/undel-bucket/k",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(restored.status(), StatusCode::OK);
+    let body = restored.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body[..], b"only");
+}
+
+#[tokio::test]
+async fn test_versioned_get_on_delete_marker_returns_method_not_allowed() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/dm-bucket", Body::empty()))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/dm-bucket?versioning")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::from(
+                    "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/dm-bucket/k",
+            Body::from("x"),
+        ))
+        .await
+        .unwrap();
+
+    let del = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            "/dm-bucket/k",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let dm_version = del
+        .headers()
+        .get("x-amz-version-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_string();
+
+    let versioned = app
+        .oneshot(signed_request(
+            Method::GET,
+            &format!("/dm-bucket/k?versionId={}", dm_version),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(versioned.status(), StatusCode::METHOD_NOT_ALLOWED);
+}
+
 #[tokio::test]
 async fn test_retention_is_enforced_when_deleting_archived_version() {
     let (app, _tmp) = test_app();
@@ -2474,9 +2939,16 @@ async fn test_retention_is_enforced_when_deleting_archived_version() {
     )
     .unwrap();
     let archived_version_id = list_body
-        .split("<VersionId>")
-        .filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
-        .find(|id| *id != "null")
+        .split("<Version>")
+        .skip(1)
+        .find(|block| block.contains("<IsLatest>false</IsLatest>"))
+        .and_then(|block| {
+            block
+                .split("<VersionId>")
+                .nth(1)
+                .and_then(|s| s.split_once("</VersionId>").map(|(id, _)| id))
+        })
+        .filter(|id| *id != "null")
         .expect("archived version id")
         .to_string();
 
@@ -2562,6 +3034,132 @@ async fn test_put_object_validates_content_md5() {
     assert_eq!(good_resp.status(), StatusCode::OK);
 }
 
+#[tokio::test]
+async fn test_x_amz_content_sha256_mismatch_returns_bad_digest() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/sha256-bucket", Body::empty()))
+        .await
+        .unwrap();
+
+    let bad_resp = app
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/sha256-bucket/object.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header(
+                    "x-amz-content-sha256",
+                    "0000000000000000000000000000000000000000000000000000000000000000",
+                )
+                .body(Body::from("hello"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(bad_resp.status(), StatusCode::BAD_REQUEST);
+    let bad_body = String::from_utf8(
+        bad_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(bad_body.contains("<Code>BadDigest</Code>"));
+    assert!(bad_body.contains("x-amz-content-sha256"));
+}
+
+#[tokio::test]
+async fn test_max_keys_zero_respects_marker_and_v2_cursors() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/cursor-bucket", Body::empty()))
+        .await
+        .unwrap();
+    for key in ["a.txt", "b.txt"] {
+        app.clone()
+            .oneshot(signed_request(
+                Method::PUT,
+                &format!("/cursor-bucket/{}", key),
+                Body::from(key.to_string()),
+            ))
+            .await
+            .unwrap();
+    }
+
+    let marker_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/cursor-bucket?max-keys=0&marker=b.txt",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let marker_body = String::from_utf8(
+        marker_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(marker_body.contains("<IsTruncated>false</IsTruncated>"));
+
+    let start_after_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/cursor-bucket?list-type=2&max-keys=0&start-after=b.txt",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let start_after_body = String::from_utf8(
+        start_after_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(start_after_body.contains("<IsTruncated>false</IsTruncated>"));
+
+    let token = URL_SAFE.encode("b.txt");
+    let token_resp = app
+        .oneshot(signed_request(
+            Method::GET,
+            &format!(
+                "/cursor-bucket?list-type=2&max-keys=0&continuation-token={}",
+                token
+            ),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    let token_body = String::from_utf8(
+        token_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(token_body.contains("<IsTruncated>false</IsTruncated>"));
+}
+
 #[tokio::test]
 async fn test_put_object_tagging_and_standard_headers_are_persisted() {
     let (app, _tmp) = test_app();

crates/myfsio-storage/src/error.rs

@@ -17,10 +17,18 @@ pub enum StorageError {
         key: String,
         version_id: String,
     },
+    #[error("Object is a delete marker: {bucket}/{key}")]
+    DeleteMarker {
+        bucket: String,
+        key: String,
+        version_id: String,
+    },
     #[error("Invalid bucket name: {0}")]
     InvalidBucketName(String),
     #[error("Invalid object key: {0}")]
     InvalidObjectKey(String),
+    #[error("Method not allowed: {0}")]
+    MethodNotAllowed(String),
     #[error("Upload not found: {0}")]
     UploadNotFound(String),
     #[error("Quota exceeded: {0}")]
@@ -42,7 +50,7 @@ impl From<StorageError> for S3Error {
                 S3Error::from_code(S3ErrorCode::NoSuchBucket).with_resource(format!("/{}", name))
             }
             StorageError::BucketAlreadyExists(name) => {
-                S3Error::from_code(S3ErrorCode::BucketAlreadyExists)
+                S3Error::from_code(S3ErrorCode::BucketAlreadyOwnedByYou)
                     .with_resource(format!("/{}", name))
             }
             StorageError::BucketNotEmpty(name) => {
@@ -58,10 +66,17 @@ impl From<StorageError> for S3Error {
                 version_id,
             } => S3Error::from_code(S3ErrorCode::NoSuchVersion)
                 .with_resource(format!("/{}/{}?versionId={}", bucket, key, version_id)),
+            StorageError::DeleteMarker {
+                bucket,
+                key,
+                version_id,
+            } => S3Error::from_code(S3ErrorCode::MethodNotAllowed)
+                .with_resource(format!("/{}/{}?versionId={}", bucket, key, version_id)),
             StorageError::InvalidBucketName(msg) => {
                 S3Error::new(S3ErrorCode::InvalidBucketName, msg)
             }
             StorageError::InvalidObjectKey(msg) => S3Error::new(S3ErrorCode::InvalidKey, msg),
+            StorageError::MethodNotAllowed(msg) => S3Error::new(S3ErrorCode::MethodNotAllowed, msg),
             StorageError::UploadNotFound(id) => S3Error::new(
                 S3ErrorCode::NoSuchUpload,
                 format!("Upload {} not found", id),

crates/myfsio-storage/src/traits.rs

@@ -62,14 +62,14 @@ pub trait StorageEngine: Send + Sync {
         version_id: &str,
     ) -> StorageResult<HashMap<String, String>>;
 
-    async fn delete_object(&self, bucket: &str, key: &str) -> StorageResult<()>;
+    async fn delete_object(&self, bucket: &str, key: &str) -> StorageResult<DeleteOutcome>;
 
     async fn delete_object_version(
         &self,
         bucket: &str,
         key: &str,
         version_id: &str,
-    ) -> StorageResult<()>;
+    ) -> StorageResult<DeleteOutcome>;
 
     async fn copy_object(
         &self,

crates/myfsio-storage/src/validation.rs

@@ -47,6 +47,7 @@ pub fn validate_object_key(
         normalized.split('/').collect()
     };
 
+
     for part in &parts {
         if part.is_empty() {
             continue;
@@ -60,6 +61,12 @@ pub fn validate_object_key(
             return Some("Object key contains invalid segments".to_string());
         }
 
+        if part.len() > 255 {
+            return Some(
+                "Object key contains a path segment that exceeds 255 bytes".to_string(),
+            );
+        }
+
         if part.chars().any(|c| (c as u32) < 32) {
             return Some("Object key contains control characters".to_string());
         }
@@ -98,6 +105,15 @@ pub fn validate_object_key(
         }
     }
 
+    for part in &non_empty_parts {
+        if *part == ".__myfsio_dirobj__"
+            || *part == ".__myfsio_empty__"
+            || part.starts_with("_index.json")
+        {
+            return Some("Object key segment uses a reserved internal name".to_string());
+        }
+    }
+
     None
 }
 
@@ -132,6 +148,13 @@ pub fn validate_bucket_name(bucket_name: &str) -> Option<String> {
         return Some("Bucket name must not be formatted as an IP address".to_string());
     }
 
+    if bucket_name.starts_with("xn--") {
+        return Some("Bucket name must not start with the reserved prefix 'xn--'".to_string());
+    }
+    if bucket_name.ends_with("-s3alias") || bucket_name.ends_with("--ol-s3") {
+        return Some("Bucket name must not end with a reserved suffix".to_string());
+    }
+
     None
 }
 

crates/myfsio-xml/Cargo.toml

@@ -8,3 +8,4 @@ myfsio-common = { path = "../myfsio-common" }
 quick-xml = { workspace = true }
 serde = { workspace = true }
 chrono = { workspace = true }
+percent-encoding = { workspace = true }

crates/myfsio-xml/src/request.rs

@@ -1,13 +1,13 @@
 use quick_xml::events::Event;
 use quick_xml::Reader;
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct DeleteObjectsRequest {
     pub objects: Vec<ObjectIdentifier>,
     pub quiet: bool,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct ObjectIdentifier {
     pub key: String,
     pub version_id: Option<String>,
@@ -86,6 +86,11 @@ pub fn parse_complete_multipart_upload(xml: &str) -> Result<CompleteMultipartUpl
 }
 
 pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
+    let trimmed = xml.trim();
+    if trimmed.is_empty() {
+        return Err("Request body is empty".to_string());
+    }
+
     let mut reader = Reader::from_str(xml);
     let mut result = DeleteObjectsRequest::default();
     let mut buf = Vec::new();
@@ -93,18 +98,43 @@ pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
     let mut current_key: Option<String> = None;
     let mut current_version_id: Option<String> = None;
     let mut in_object = false;
+    let mut saw_delete_root = false;
+    let mut first_element_seen = false;
 
     loop {
-        match reader.read_event_into(&mut buf) {
+        let event = reader.read_event_into(&mut buf);
+        match event {
             Ok(Event::Start(ref e)) => {
                 let name = String::from_utf8_lossy(e.name().as_ref()).to_string();
                 current_tag = name.clone();
-                if name == "Object" {
+                if !first_element_seen {
+                    first_element_seen = true;
+                    if name != "Delete" {
+                        return Err(format!(
+                            "Expected <Delete> root element, found <{}>",
+                            name
+                        ));
+                    }
+                    saw_delete_root = true;
+                } else if name == "Object" {
                     in_object = true;
                     current_key = None;
                     current_version_id = None;
                 }
             }
+            Ok(Event::Empty(ref e)) => {
+                let name = String::from_utf8_lossy(e.name().as_ref()).to_string();
+                if !first_element_seen {
+                    first_element_seen = true;
+                    if name != "Delete" {
+                        return Err(format!(
+                            "Expected <Delete> root element, found <{}>",
+                            name
+                        ));
+                    }
+                    saw_delete_root = true;
+                }
+            }
             Ok(Event::Text(ref e)) => {
                 let text = e.unescape().map_err(|e| e.to_string())?.to_string();
                 match current_tag.as_str() {
@@ -139,6 +169,13 @@ pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
         buf.clear();
     }
 
+    if !saw_delete_root {
+        return Err("Expected <Delete> root element".to_string());
+    }
+    if result.objects.is_empty() {
+        return Err("Delete request must contain at least one <Object>".to_string());
+    }
+
     Ok(result)
 }
 

crates/myfsio-xml/src/response.rs

@@ -8,10 +8,21 @@ pub fn format_s3_datetime(dt: &DateTime<Utc>) -> String {
     dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
 }
 
-pub fn rate_limit_exceeded_xml() -> String {
-    "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
-<Error><Code>SlowDown</Code><Message>Rate limit exceeded</Message></Error>"
-        .to_string()
+pub fn rate_limit_exceeded_xml(resource: &str, request_id: &str) -> String {
+    format!(
+        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
+<Error><Code>SlowDown</Code><Message>Please reduce your request rate</Message><Resource>{}</Resource><RequestId>{}</RequestId></Error>",
+        xml_escape(resource),
+        xml_escape(request_id),
+    )
+}
+
+fn xml_escape(s: &str) -> String {
+    s.replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+        .replace('"', "&quot;")
+        .replace('\'', "&apos;")
 }
 
 pub fn list_buckets_xml(owner_id: &str, owner_name: &str, buckets: &[BucketMeta]) -> String {
@@ -62,6 +73,21 @@ pub fn list_buckets_xml(owner_id: &str, owner_name: &str, buckets: &[BucketMeta]
     String::from_utf8(writer.into_inner().into_inner()).unwrap()
 }
 
+fn maybe_url_encode(value: &str, encoding_type: Option<&str>) -> String {
+    if matches!(encoding_type, Some(v) if v.eq_ignore_ascii_case("url")) {
+        percent_encoding::utf8_percent_encode(value, KEY_ENCODE_SET).to_string()
+    } else {
+        value.to_string()
+    }
+}
+
+const KEY_ENCODE_SET: &percent_encoding::AsciiSet = &percent_encoding::NON_ALPHANUMERIC
+    .remove(b'-')
+    .remove(b'_')
+    .remove(b'.')
+    .remove(b'~')
+    .remove(b'/');
+
 pub fn list_objects_v2_xml(
     bucket_name: &str,
     prefix: &str,
@@ -73,6 +99,34 @@ pub fn list_objects_v2_xml(
     continuation_token: Option<&str>,
     next_continuation_token: Option<&str>,
     key_count: usize,
+) -> String {
+    list_objects_v2_xml_with_encoding(
+        bucket_name,
+        prefix,
+        delimiter,
+        max_keys,
+        objects,
+        common_prefixes,
+        is_truncated,
+        continuation_token,
+        next_continuation_token,
+        key_count,
+        None,
+    )
+}
+
+pub fn list_objects_v2_xml_with_encoding(
+    bucket_name: &str,
+    prefix: &str,
+    delimiter: &str,
+    max_keys: usize,
+    objects: &[ObjectMeta],
+    common_prefixes: &[String],
+    is_truncated: bool,
+    continuation_token: Option<&str>,
+    next_continuation_token: Option<&str>,
+    key_count: usize,
+    encoding_type: Option<&str>,
 ) -> String {
     let mut writer = Writer::new(Cursor::new(Vec::new()));
 
@@ -85,13 +139,22 @@ pub fn list_objects_v2_xml(
     writer.write_event(Event::Start(start)).unwrap();
 
     write_text_element(&mut writer, "Name", bucket_name);
-    write_text_element(&mut writer, "Prefix", prefix);
+    write_text_element(&mut writer, "Prefix", &maybe_url_encode(prefix, encoding_type));
     if !delimiter.is_empty() {
-        write_text_element(&mut writer, "Delimiter", delimiter);
+        write_text_element(
+            &mut writer,
+            "Delimiter",
+            &maybe_url_encode(delimiter, encoding_type),
+        );
     }
     write_text_element(&mut writer, "MaxKeys", &max_keys.to_string());
     write_text_element(&mut writer, "KeyCount", &key_count.to_string());
     write_text_element(&mut writer, "IsTruncated", &is_truncated.to_string());
+    if let Some(encoding) = encoding_type {
+        if !encoding.is_empty() {
+            write_text_element(&mut writer, "EncodingType", encoding);
+        }
+    }
 
     if let Some(token) = continuation_token {
         write_text_element(&mut writer, "ContinuationToken", token);
@@ -104,7 +167,7 @@ pub fn list_objects_v2_xml(
         writer
             .write_event(Event::Start(BytesStart::new("Contents")))
             .unwrap();
-        write_text_element(&mut writer, "Key", &obj.key);
+        write_text_element(&mut writer, "Key", &maybe_url_encode(&obj.key, encoding_type));
         write_text_element(
             &mut writer,
             "LastModified",
@@ -128,7 +191,7 @@ pub fn list_objects_v2_xml(
         writer
             .write_event(Event::Start(BytesStart::new("CommonPrefixes")))
             .unwrap();
-        write_text_element(&mut writer, "Prefix", prefix);
+        write_text_element(&mut writer, "Prefix", &maybe_url_encode(prefix, encoding_type));
         writer
             .write_event(Event::End(BytesEnd::new("CommonPrefixes")))
             .unwrap();
@@ -151,6 +214,32 @@ pub fn list_objects_v1_xml(
     common_prefixes: &[String],
     is_truncated: bool,
     next_marker: Option<&str>,
+) -> String {
+    list_objects_v1_xml_with_encoding(
+        bucket_name,
+        prefix,
+        marker,
+        delimiter,
+        max_keys,
+        objects,
+        common_prefixes,
+        is_truncated,
+        next_marker,
+        None,
+    )
+}
+
+pub fn list_objects_v1_xml_with_encoding(
+    bucket_name: &str,
+    prefix: &str,
+    marker: &str,
+    delimiter: &str,
+    max_keys: usize,
+    objects: &[ObjectMeta],
+    common_prefixes: &[String],
+    is_truncated: bool,
+    next_marker: Option<&str>,
+    encoding_type: Option<&str>,
 ) -> String {
     let mut writer = Writer::new(Cursor::new(Vec::new()));
 
@@ -163,27 +252,36 @@ pub fn list_objects_v1_xml(
     writer.write_event(Event::Start(start)).unwrap();
 
     write_text_element(&mut writer, "Name", bucket_name);
-    write_text_element(&mut writer, "Prefix", prefix);
-    write_text_element(&mut writer, "Marker", marker);
+    write_text_element(&mut writer, "Prefix", &maybe_url_encode(prefix, encoding_type));
+    write_text_element(&mut writer, "Marker", &maybe_url_encode(marker, encoding_type));
     write_text_element(&mut writer, "MaxKeys", &max_keys.to_string());
     write_text_element(&mut writer, "IsTruncated", &is_truncated.to_string());
 
     if !delimiter.is_empty() {
-        write_text_element(&mut writer, "Delimiter", delimiter);
+        write_text_element(
+            &mut writer,
+            "Delimiter",
+            &maybe_url_encode(delimiter, encoding_type),
+        );
     }
     if !delimiter.is_empty() && is_truncated {
         if let Some(nm) = next_marker {
             if !nm.is_empty() {
-                write_text_element(&mut writer, "NextMarker", nm);
+                write_text_element(&mut writer, "NextMarker", &maybe_url_encode(nm, encoding_type));
             }
         }
     }
+    if let Some(encoding) = encoding_type {
+        if !encoding.is_empty() {
+            write_text_element(&mut writer, "EncodingType", encoding);
+        }
+    }
 
     for obj in objects {
         writer
             .write_event(Event::Start(BytesStart::new("Contents")))
             .unwrap();
-        write_text_element(&mut writer, "Key", &obj.key);
+        write_text_element(&mut writer, "Key", &maybe_url_encode(&obj.key, encoding_type));
         write_text_element(
             &mut writer,
             "LastModified",
@@ -202,7 +300,7 @@ pub fn list_objects_v1_xml(
         writer
             .write_event(Event::Start(BytesStart::new("CommonPrefixes")))
             .unwrap();
-        write_text_element(&mut writer, "Prefix", cp);
+        write_text_element(&mut writer, "Prefix", &maybe_url_encode(cp, encoding_type));
         writer
             .write_event(Event::End(BytesEnd::new("CommonPrefixes")))
             .unwrap();
@@ -325,8 +423,15 @@ pub fn copy_object_result_xml(etag: &str, last_modified: &str) -> String {
     String::from_utf8(writer.into_inner().into_inner()).unwrap()
 }
 
+pub struct DeletedEntry {
+    pub key: String,
+    pub version_id: Option<String>,
+    pub delete_marker: bool,
+    pub delete_marker_version_id: Option<String>,
+}
+
 pub fn delete_result_xml(
-    deleted: &[(String, Option<String>)],
+    deleted: &[DeletedEntry],
     errors: &[(String, String, String)],
     quiet: bool,
 ) -> String {
@@ -340,14 +445,20 @@ pub fn delete_result_xml(
     writer.write_event(Event::Start(start)).unwrap();
 
     if !quiet {
-        for (key, version_id) in deleted {
+        for entry in deleted {
             writer
                 .write_event(Event::Start(BytesStart::new("Deleted")))
                 .unwrap();
-            write_text_element(&mut writer, "Key", key);
-            if let Some(vid) = version_id {
+            write_text_element(&mut writer, "Key", &entry.key);
+            if let Some(ref vid) = entry.version_id {
                 write_text_element(&mut writer, "VersionId", vid);
             }
+            if entry.delete_marker {
+                write_text_element(&mut writer, "DeleteMarker", "true");
+                if let Some(ref dm_vid) = entry.delete_marker_version_id {
+                    write_text_element(&mut writer, "DeleteMarkerVersionId", dm_vid);
+                }
+            }
             writer
                 .write_event(Event::End(BytesEnd::new("Deleted")))
                 .unwrap();