test(kms): use absolute fixture paths on Windows

* fix(s3): normalize leading-slash keys in bulk deletes

* test(s3): align bulk delete reserved-prefix guard expectations

crates/checksums/src/lib.rs

@@ -902,4 +902,104 @@ mod tests {
         assert_eq!(raw.len(), 8);
         assert_eq!(&raw[..], reference.digest().to_be_bytes().as_slice());
     }
+
+    #[test]
+    fn checksums_streaming_chunks_and_header_encoding_match_oneshot_across_unaligned_boundaries() {
+        use crate::http::HttpChecksum;
+
+        let all = [
+            ChecksumAlgorithm::Crc32,
+            ChecksumAlgorithm::Crc32c,
+            ChecksumAlgorithm::Crc64Nvme,
+            ChecksumAlgorithm::Sha1,
+            ChecksumAlgorithm::Sha256,
+            ChecksumAlgorithm::Sha512,
+            ChecksumAlgorithm::Xxhash3,
+            ChecksumAlgorithm::Xxhash64,
+            ChecksumAlgorithm::Xxhash128,
+        ];
+
+        for &payload_len in &[0usize, 1, 31, 33, 1024, 4096 + 37, 64 * 1024 + 19] {
+            let payload: Vec<u8> = (0..payload_len).map(|i| ((i * 19 + 5) % 253) as u8).collect();
+            for algorithm in all {
+                let mut one_shot = algorithm.into_impl();
+                one_shot.update(&payload);
+                let expected_raw = one_shot.finalize();
+                assert_eq!(expected_raw.len(), algorithm.raw_len());
+
+                let mut one_shot_hdr = algorithm.into_impl();
+                one_shot_hdr.update(&payload);
+                let expected_headers = one_shot_hdr.headers();
+
+                for &chunk_size in &[1usize, 31, 1021, 4096] {
+                    if payload_len > 4096 && chunk_size == 1 {
+                        continue;
+                    }
+                    let mut streaming: Box<dyn HttpChecksum> = algorithm.into_impl();
+                    for chunk in payload.chunks(chunk_size) {
+                        streaming.update(chunk);
+                    }
+                    let actual_headers = streaming.headers();
+                    assert_eq!(
+                        actual_headers.get(algorithm.http_header_name()),
+                        expected_headers.get(algorithm.http_header_name()),
+                        "{algorithm:?} streaming mismatch at payload_len={payload_len}, chunk_size={chunk_size}"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    #[ignore = "utility throughput microbenchmark; run via cargo test -p rustfs-checksums --lib --release -- --ignored --nocapture"]
+    fn microbench_checksums_streaming_chunks_and_header_encoding() {
+        use crate::http::HttpChecksum;
+        use std::hint::black_box;
+        use std::time::Instant;
+
+        let all = [
+            ChecksumAlgorithm::Crc32,
+            ChecksumAlgorithm::Crc32c,
+            ChecksumAlgorithm::Crc64Nvme,
+            ChecksumAlgorithm::Sha1,
+            ChecksumAlgorithm::Sha256,
+            ChecksumAlgorithm::Sha512,
+            ChecksumAlgorithm::Xxhash3,
+            ChecksumAlgorithm::Xxhash64,
+            ChecksumAlgorithm::Xxhash128,
+        ];
+
+        let payload: Vec<u8> = (0..(256 * 1024)).map(|i| ((i * 19 + 5) % 253) as u8).collect();
+        for &chunk_size in &[4 * 1024usize, 64 * 1024, 256 * 1024] {
+            for algorithm in all {
+                let mut one_shot = algorithm.into_impl();
+                one_shot.update(&payload);
+                let expected_headers = one_shot.headers();
+
+                let iters = 8usize;
+                let t0 = Instant::now();
+                let mut last_headers = http::HeaderMap::new();
+                for _ in 0..iters {
+                    let mut streaming: Box<dyn HttpChecksum> = algorithm.into_impl();
+                    for chunk in payload.chunks(chunk_size) {
+                        streaming.update(black_box(chunk));
+                    }
+                    last_headers = streaming.headers();
+                    black_box(&last_headers);
+                }
+                let elapsed = t0.elapsed();
+                assert_eq!(
+                    last_headers.get(algorithm.http_header_name()),
+                    expected_headers.get(algorithm.http_header_name()),
+                    "{algorithm:?} streaming chunk_size={chunk_size} header mismatch"
+                );
+                let total_mib = (payload.len() * iters) as f64 / (1024.0 * 1024.0);
+                let mibs = total_mib / elapsed.as_secs_f64().max(1e-9);
+                println!(
+                    "[microbench][checksums] algo={algorithm:?} chunk_size={chunk_size}B throughput={mibs:.1} MiB/s (size={}B)",
+                    algorithm.raw_len()
+                );
+            }
+        }
+    }
 }

crates/e2e_test/src/leading_slash_key_test.rs

@@ -21,12 +21,15 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::common::{RustFSTestEnvironment, init_logging};
+    use crate::common::{RustFSTestEnvironment, admin_create_user, init_logging};
     use aws_sdk_s3::Client;
     use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::{BucketVersioningStatus, Delete, ObjectIdentifier, VersioningConfiguration};
     use std::error::Error;
     use tracing::info;
 
+    type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
+
     async fn create_bucket(client: &Client, bucket: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
         client.create_bucket().bucket(bucket).send().await?;
         Ok(())
@@ -90,7 +93,7 @@ mod tests {
         Ok(())
     }
 
-    /// Duplicate and repeated slashes after a leading slash collapse MinIO-style.
+    /// Duplicate and repeated slashes after a leading slash collapse MinIO-style across single and bulk operations.
     #[tokio::test]
     async fn test_put_object_with_duplicate_slashes_normalized() -> Result<(), Box<dyn Error + Send + Sync>> {
         init_logging();
@@ -144,8 +147,258 @@ mod tests {
             "GET after DELETE with raw key must return HTTP 404, got {error:?}"
         );
 
+        assert_bulk_delete_normalizes_keys(&client).await?;
+        assert_bulk_delete_authorizes_normalized_keys(&env, &client).await?;
+        assert_bulk_delete_preserves_version_identity(&client).await?;
+
         env.stop_server();
         info!("Test completed successfully");
         Ok(())
     }
+
+    async fn assert_bulk_delete_normalizes_keys(client: &Client) -> TestResult {
+        let bucket = "test-bulk-leading-slash";
+        create_bucket(client, bucket).await?;
+        let cases = [
+            ("/foo/bar", "foo/bar"),
+            ("//keyname", "keyname"),
+            ("/dir///sub//file", "dir/sub/file"),
+            ("/folder///", "folder/"),
+            ("/literal%2F+名", "literal%2F+名"),
+            ("plain", "plain"),
+        ];
+        let invalid = ["///", "/dir/../keep", "dir//keep"];
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("keep")
+            .body(ByteStream::from_static(b"keep"))
+            .send()
+            .await?;
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("dir/keep")
+            .body(ByteStream::from_static(b"keep"))
+            .send()
+            .await?;
+
+        for quiet in [false, true] {
+            for (raw, _) in cases {
+                client
+                    .put_object()
+                    .bucket(bucket)
+                    .key(raw)
+                    .body(ByteStream::from_static(b""))
+                    .send()
+                    .await?;
+            }
+            let listing = client.list_objects_v2().bucket(bucket).send().await?;
+            let stored: Vec<_> = listing.contents().iter().filter_map(|object| object.key()).collect();
+            let mut expected: Vec<_> = cases
+                .iter()
+                .map(|(_, normalized)| *normalized)
+                .chain(["dir/keep", "keep"])
+                .collect();
+            expected.sort_unstable();
+            assert_eq!(stored, expected, "PUT must store the exact normalized targets before bulk deletion");
+            let objects = cases
+                .iter()
+                .map(|(raw, _)| *raw)
+                .chain(invalid)
+                .map(|key| ObjectIdentifier::builder().key(key).build())
+                .collect::<Result<Vec<_>, _>>()?;
+            let delete = Delete::builder().set_objects(Some(objects)).quiet(quiet).build()?;
+            // A second identical batch also pins idempotent deletion of absent normalized keys.
+            for _ in 0..2 {
+                let output = client.delete_objects().bucket(bucket).delete(delete.clone()).send().await?;
+                let errors: Vec<_> = output.errors().iter().map(|error| (error.key(), error.code())).collect();
+                assert_eq!(
+                    errors,
+                    invalid.map(|key| (Some(key), Some("InvalidArgument"))),
+                    "unexpected per-key errors: {output:?}"
+                );
+                let deleted: Vec<_> = output.deleted().iter().filter_map(|object| object.key()).collect();
+                let expected = if quiet {
+                    vec![]
+                } else {
+                    cases.iter().map(|(raw, _)| *raw).collect()
+                };
+                assert_eq!(deleted, expected, "responses must retain request keys, including directory slashes");
+            }
+            let listing = client.list_objects_v2().bucket(bucket).send().await?;
+            let remaining: Vec<_> = listing.contents().iter().filter_map(|object| object.key()).collect();
+            assert_eq!(
+                remaining,
+                ["dir/keep", "keep"],
+                "bulk delete must remove normalized objects without resolving dot segments"
+            );
+        }
+        Ok(())
+    }
+
+    async fn assert_bulk_delete_authorizes_normalized_keys(env: &RustFSTestEnvironment, admin: &Client) -> TestResult {
+        let bucket = "test-bulk-slash-policy";
+        let user = "bulk-slash-user";
+        admin_create_user(env, user, "bulkSlashTestSecret123!").await?;
+        create_bucket(admin, bucket).await?;
+        for key in ["allowed/item", "protected/item"] {
+            admin
+                .put_object()
+                .bucket(bucket)
+                .key(key)
+                .body(ByteStream::from_static(b"keep"))
+                .send()
+                .await?;
+        }
+        let policy = serde_json::json!({
+            "Version": "2012-10-17",
+            "Statement": [
+                {"Effect": "Allow", "Principal": {"AWS": [user]}, "Action": ["s3:DeleteObject"],
+                 "Resource": [format!("arn:aws:s3:::{bucket}/*")]},
+                {"Effect": "Deny", "Principal": {"AWS": [user]}, "Action": ["s3:DeleteObject"],
+                 "Resource": [format!("arn:aws:s3:::{bucket}/protected/*")]}
+            ]
+        });
+        admin
+            .put_bucket_policy()
+            .bucket(bucket)
+            .policy(policy.to_string())
+            .send()
+            .await?;
+        let client = env.create_s3_client_with_credentials(user, "bulkSlashTestSecret123!");
+        let output = client
+            .delete_objects()
+            .bucket(bucket)
+            .delete(
+                Delete::builder()
+                    .objects(ObjectIdentifier::builder().key("/allowed///item").build()?)
+                    .objects(ObjectIdentifier::builder().key("/protected///item").build()?)
+                    .build()?,
+            )
+            .send()
+            .await?;
+        assert_eq!(output.deleted().len(), 1, "allowed normalized key must be deleted: {output:?}");
+        assert_eq!(output.deleted()[0].key(), Some("/allowed///item"));
+        assert_eq!(output.errors().len(), 1, "normalized resource denial must apply: {output:?}");
+        assert_eq!(output.errors()[0].key(), Some("/protected///item"));
+        assert_eq!(output.errors()[0].code(), Some("AccessDenied"));
+        let listing = admin.list_objects_v2().bucket(bucket).send().await?;
+        let remaining: Vec<_> = listing.contents().iter().filter_map(|object| object.key()).collect();
+        assert_eq!(remaining, ["protected/item"], "only the authorized object may be removed");
+        let protected = admin.get_object().bucket(bucket).key("protected/item").send().await?;
+        assert_eq!(protected.body.collect().await?.into_bytes().as_ref(), b"keep");
+        Ok(())
+    }
+
+    async fn assert_bulk_delete_preserves_version_identity(client: &Client) -> TestResult {
+        let bucket = "test-bulk-slash-versions";
+        let raw = "/history///file";
+        let normalized = "history/file";
+        create_bucket(client, bucket).await?;
+        client
+            .put_bucket_versioning()
+            .bucket(bucket)
+            .versioning_configuration(
+                VersioningConfiguration::builder()
+                    .status(BucketVersioningStatus::Enabled)
+                    .build(),
+            )
+            .send()
+            .await?;
+        let first = client
+            .put_object()
+            .bucket(bucket)
+            .key(raw)
+            .body(ByteStream::from_static(b"old"))
+            .send()
+            .await?;
+        let first_id = first.version_id().ok_or("first PUT omitted version ID")?;
+        let latest = client
+            .put_object()
+            .bucket(bucket)
+            .key(raw)
+            .body(ByteStream::from_static(b"latest"))
+            .send()
+            .await?;
+        let latest_id = latest.version_id().ok_or("latest PUT omitted version ID")?;
+        let invalid = client
+            .delete_objects()
+            .bucket(bucket)
+            .delete(
+                Delete::builder()
+                    .objects(ObjectIdentifier::builder().key("///").build()?)
+                    .build()?,
+            )
+            .send()
+            .await?;
+        assert!(invalid.deleted().is_empty(), "slash-only keys must not delete a versioned object");
+        assert_eq!(
+            invalid.errors().len(),
+            1,
+            "slash-only keys must remain invalid in versioned buckets: {invalid:?}"
+        );
+        assert_eq!(invalid.errors()[0].key(), Some("///"));
+        assert_eq!(invalid.errors()[0].code(), Some("InvalidArgument"));
+        let deleted = client
+            .delete_objects()
+            .bucket(bucket)
+            .delete(
+                Delete::builder()
+                    .objects(ObjectIdentifier::builder().key(raw).version_id(first_id).build()?)
+                    .build()?,
+            )
+            .send()
+            .await?;
+        assert!(deleted.errors().is_empty(), "historical delete failed: {deleted:?}");
+        assert_eq!(deleted.deleted().len(), 1);
+        assert_eq!(deleted.deleted()[0].key(), Some(raw));
+        assert_eq!(deleted.deleted()[0].version_id(), Some(first_id));
+        let listing = client.list_object_versions().bucket(bucket).send().await?;
+        assert_eq!(listing.versions().len(), 1, "only the latest data version should remain: {listing:?}");
+        assert_eq!(listing.versions()[0].version_id(), Some(latest_id));
+
+        let deleted = client
+            .delete_objects()
+            .bucket(bucket)
+            .delete(
+                Delete::builder()
+                    .objects(ObjectIdentifier::builder().key(raw).build()?)
+                    .build()?,
+            )
+            .send()
+            .await?;
+        assert!(deleted.errors().is_empty(), "delete marker creation failed: {deleted:?}");
+        assert_eq!(deleted.deleted().len(), 1);
+        assert_eq!(deleted.deleted()[0].key(), Some(raw));
+        assert_eq!(deleted.deleted()[0].delete_marker(), Some(true));
+        let marker = deleted.deleted()[0]
+            .delete_marker_version_id()
+            .ok_or("missing delete marker ID")?;
+        let listing = client.list_object_versions().bucket(bucket).send().await?;
+        assert_eq!(listing.delete_markers().len(), 1);
+        assert_eq!(listing.delete_markers()[0].key(), Some(normalized));
+        assert_eq!(listing.delete_markers()[0].version_id(), Some(marker));
+
+        let deleted = client
+            .delete_objects()
+            .bucket(bucket)
+            .delete(
+                Delete::builder()
+                    .objects(ObjectIdentifier::builder().key(raw).version_id(marker).build()?)
+                    .build()?,
+            )
+            .send()
+            .await?;
+        assert!(deleted.errors().is_empty(), "explicit marker removal failed: {deleted:?}");
+        assert_eq!(deleted.deleted().len(), 1);
+        assert_eq!(deleted.deleted()[0].key(), Some(raw));
+        assert_eq!(deleted.deleted()[0].version_id(), Some(marker));
+        let listing = client.list_object_versions().bucket(bucket).send().await?;
+        assert!(listing.delete_markers().is_empty());
+        let restored = client.get_object().bucket(bucket).key(normalized).send().await?;
+        assert_eq!(restored.version_id(), Some(latest_id));
+        assert_eq!(restored.body.collect().await?.into_bytes().as_ref(), b"latest");
+        Ok(())
+    }
 }

crates/kms/src/api_types.rs

@@ -917,9 +917,10 @@ mod tests {
 
     #[test]
     fn test_configure_request_development_defaults_require_opt_in() {
+        let key_dir = std::env::temp_dir().join("kms-key-dir");
         let local_raw = serde_json::json!({
             "backend_type": "local",
-            "key_dir": "/tmp/kms-key-dir"
+            "key_dir": key_dir
         });
         let request: ConfigureKmsRequest = serde_json::from_value(local_raw).expect("local request should deserialize");
         let config = request.to_kms_config();
@@ -927,7 +928,7 @@ mod tests {
 
         let local_opt_in_raw = serde_json::json!({
             "backend_type": "local",
-            "key_dir": "/tmp/kms-key-dir",
+            "key_dir": key_dir,
             "allow_insecure_dev_defaults": true
         });
         let request: ConfigureKmsRequest = serde_json::from_value(local_opt_in_raw).expect("local request should deserialize");

crates/kms/src/config.rs

@@ -1608,10 +1608,14 @@ mod tests {
         assert!(config.validate().is_err());
         assert!(config.with_insecure_development_defaults().validate().is_ok());
 
+        #[cfg(windows)]
+        let production_key_dir = PathBuf::from(r"C:\rustfs\kms");
+        #[cfg(not(windows))]
+        let production_key_dir = PathBuf::from("/var/lib/rustfs/kms");
         let production_config = KmsConfig {
             backend: KmsBackend::Local,
             backend_config: BackendConfig::Local(LocalConfig {
-                key_dir: PathBuf::from("/var/lib/rustfs/kms"),
+                key_dir: production_key_dir,
                 master_key: Some("production-master-key".to_string()),
                 file_permissions: Some(0o600),
             }),

crates/kms/tests/behavior_lifecycle.rs

@@ -336,7 +336,11 @@ async fn invalid_configurations_are_rejected_before_anything_starts() {
     assert_configuration_error(relative.validate(), "must be an absolute path");
 
     // Production-shaped Local config: no master key is refused without opt-in.
-    let production_without_master_key = KmsConfig::local("/var/lib/rustfs/kms-behavior".into());
+    #[cfg(windows)]
+    let production_key_dir = r"C:\rustfs\kms-behavior";
+    #[cfg(not(windows))]
+    let production_key_dir = "/var/lib/rustfs/kms-behavior";
+    let production_without_master_key = KmsConfig::local(production_key_dir.into());
     assert_configuration_error(production_without_master_key.validate(), "requires a master key");
 
     // ...and so is a key directory under the process temp dir.

crates/rio/src/checksum.rs

@@ -1873,4 +1873,123 @@ mod tests {
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
         assert!(err.to_string().contains("invalid checksum value"), "unexpected error: {err}");
     }
+
+    fn software_crc32_bitwise(reflected_poly: u32, data: &[u8]) -> [u8; 4] {
+        let mut crc = !0u32;
+        for &b in data {
+            crc ^= u32::from(b);
+            for _ in 0..8 {
+                let mask = (crc & 1).wrapping_neg();
+                crc = (crc >> 1) ^ (reflected_poly & mask);
+            }
+        }
+        (!crc).to_be_bytes()
+    }
+
+    fn software_crc64_nvme_bitwise(data: &[u8]) -> [u8; 8] {
+        let reflected_poly = super::CRC64_NVME_POLYNOMIAL.reverse_bits();
+        let mut crc = !0u64;
+        for &b in data {
+            crc ^= u64::from(b);
+            for _ in 0..8 {
+                let mask = (crc & 1).wrapping_neg();
+                crc = (crc >> 1) ^ (reflected_poly & mask);
+            }
+        }
+        (!crc).to_be_bytes()
+    }
+
+    #[test]
+    fn rio_crc_simd_matches_bitwise_scalar_across_tail_and_vector_boundaries() {
+        for &size in &[
+            0usize,
+            1,
+            7,
+            8,
+            15,
+            16,
+            17,
+            31,
+            32,
+            33,
+            63,
+            64,
+            65,
+            127,
+            128,
+            129,
+            255,
+            256,
+            257,
+            1024,
+            4096 + 17,
+            64 * 1024,
+        ] {
+            let payload: Vec<u8> = (0..size).map(|i| ((i * 31 + 17) % 251) as u8).collect();
+            let crc32_simd = Checksum::new_from_data(ChecksumType::CRC32, &payload).expect("crc32");
+            let crc32c_simd = Checksum::new_from_data(ChecksumType::CRC32C, &payload).expect("crc32c");
+            let crc64_simd = Checksum::new_from_data(ChecksumType::CRC64_NVME, &payload).expect("crc64nvme");
+
+            let crc32_sw = software_crc32_bitwise(0xEDB88320, &payload);
+            let crc32c_sw = software_crc32_bitwise(0x82F63B78, &payload);
+            let crc64_sw = software_crc64_nvme_bitwise(&payload);
+
+            assert_eq!(crc32_simd.raw.as_slice(), &crc32_sw, "CRC32 SIMD vs scalar mismatch at {size}B");
+            assert_eq!(crc32c_simd.raw.as_slice(), &crc32c_sw, "CRC32C SIMD vs scalar mismatch at {size}B");
+            assert_eq!(crc64_simd.raw.as_slice(), &crc64_sw, "CRC64_NVME SIMD vs scalar mismatch at {size}B");
+        }
+    }
+
+    #[test]
+    #[ignore = "utility throughput microbenchmark; run via cargo test -p rustfs-rio --lib --release -- --ignored --nocapture"]
+    fn microbench_rio_checksum_simd_vs_scalar_and_all_base_algorithms() {
+        use std::hint::black_box;
+        use std::io::Write as _;
+        use std::time::Instant;
+
+        for &(size, iters) in &[(1024usize, 100usize), (64 * 1024, 20), (1024 * 1024, 4)] {
+            let payload: Vec<u8> = (0..size).map(|i| ((i * 31 + 17) % 251) as u8).collect();
+            let total_mib = (size * iters) as f64 / (1024.0 * 1024.0);
+
+            // Verify SIMD-accelerated CRC32 / CRC32C / CRC64_NVME against bitwise scalar reference at 1 KiB & 64 KiB.
+            if size <= 64 * 1024 {
+                let crc32_simd = Checksum::new_from_data(ChecksumType::CRC32, &payload).expect("crc32");
+                let crc32c_simd = Checksum::new_from_data(ChecksumType::CRC32C, &payload).expect("crc32c");
+                let crc64_simd = Checksum::new_from_data(ChecksumType::CRC64_NVME, &payload).expect("crc64nvme");
+
+                let t_sw = Instant::now();
+                let crc32_sw = software_crc32_bitwise(0xEDB88320, black_box(&payload));
+                let crc32c_sw = software_crc32_bitwise(0x82F63B78, black_box(&payload));
+                let crc64_sw = software_crc64_nvme_bitwise(black_box(&payload));
+                let sw_elapsed = t_sw.elapsed();
+
+                assert_eq!(crc32_simd.raw.as_slice(), &crc32_sw, "CRC32 SIMD vs scalar mismatch at {size}B");
+                assert_eq!(crc32c_simd.raw.as_slice(), &crc32c_sw, "CRC32C SIMD vs scalar mismatch at {size}B");
+                assert_eq!(crc64_simd.raw.as_slice(), &crc64_sw, "CRC64_NVME SIMD vs scalar mismatch at {size}B");
+                println!(
+                    "[microbench][rio::checksum] size={size}B bitwise scalar reference check passed in {:?}",
+                    sw_elapsed
+                );
+            }
+
+            for &checksum_type in super::BASE_CHECKSUM_TYPES {
+                let mut hasher = checksum_type.hasher().expect("base checksum type must have hasher");
+                let t0 = Instant::now();
+                let mut digest = Vec::new();
+                for _ in 0..iters {
+                    hasher.reset();
+                    hasher.write_all(black_box(&payload)).expect("hasher write_all");
+                    digest = hasher.finalize();
+                    black_box(&digest);
+                }
+                let elapsed = t0.elapsed();
+                assert_eq!(digest.len(), checksum_type.raw_byte_len());
+                let mibs = total_mib / elapsed.as_secs_f64().max(1e-9);
+                println!(
+                    "[microbench][rio::checksum] size={size}B algo={} throughput={mibs:.1} MiB/s",
+                    checksum_type
+                );
+            }
+        }
+    }
 }

crates/scanner/src/scanner_folder.rs

@@ -425,10 +425,7 @@ enum FolderScanSource {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FolderScanMode {
     Namespace,
-    ErasureCandidate {
-        reuse_namespace: bool,
-        parent_has_metadata: bool,
-    },
+    ErasureCandidate { reuse_namespace: bool },
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -2068,9 +2065,7 @@ impl FolderScanner {
                     if candidate_unclassified {
                         // Apply namespace reuse only after the current listing
                         // proves there is no direct erasure part.
-                        if let FolderScanMode::ErasureCandidate {
-                            reuse_namespace: true, ..
-                        } = mode
+                        if let FolderScanMode::ErasureCandidate { reuse_namespace: true } = mode
                             && !into.compacted
                             && self.old_cache.find(&folder.name).is_some()
                         {
@@ -2084,23 +2079,10 @@ impl FolderScanner {
                             return Ok(FolderScanOutcome::Namespace);
                         }
                         candidate_unclassified = false;
-                        // An empty UUID directory beside object metadata has
-                        // no namespace work. Its listing was already charged;
-                        // nonempty namespaces still need directory admission.
-                        let empty_object_child = matches!(
-                            mode,
-                            FolderScanMode::ErasureCandidate {
-                                parent_has_metadata: true,
-                                ..
-                            }
-                        ) && deferred_metadata.is_none()
-                            && queued_folders.is_empty();
-                        if !empty_object_child {
-                            if !self.budget.try_start_directory() {
-                                return Err(ScannerError::Other("Operation cancelled".to_string()));
-                            }
-                            wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
+                        if !self.budget.try_start_directory() {
+                            return Err(ScannerError::Other("Operation cancelled".to_string()));
                         }
+                        wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                     }
                     let Some(entry) = deferred_metadata.take() else { break };
                     (entry, true)
@@ -2570,6 +2552,20 @@ impl FolderScanner {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
 
+            if found_object_metadata && self.is_erasure_mode {
+                // If we found an object in erasure mode, we skip subdirs (only datadirs)...
+                debug!(
+                    target: "rustfs::scanner::folder",
+                    event = EVENT_SCANNER_FOLDER_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_FOLDER,
+                    folder = %folder.name,
+                    state = "erasure_object_found",
+                    "Scanner folder descent stopped after erasure object"
+                );
+                break;
+            }
+
             // Checkpointed sweeps need child entries to validate their resume
             // frontier (#8292). Outside that, also compact a wide bucket root:
             // excluding it left #8961's ~170k top-level prefixes unbounded until
@@ -2658,7 +2654,6 @@ impl FolderScanner {
                 let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
                     FolderScanMode::ErasureCandidate {
                         reuse_namespace: reuse_covered,
-                        parent_has_metadata: found_object_metadata,
                     }
                 } else {
                     FolderScanMode::Namespace
@@ -2726,10 +2721,7 @@ impl FolderScanner {
                                 }
                                 // Sampling may skip namespace work, never the
                                 // classification that protects its parent object.
-                                child_mode = FolderScanMode::ErasureCandidate {
-                                    reuse_namespace: true,
-                                    parent_has_metadata: found_object_metadata,
-                                };
+                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
                             } else {
                                 folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                             }
@@ -2824,11 +2816,6 @@ impl FolderScanner {
                         updates.cache.delete_recursive(&h);
                     }
                     into.children.remove(&clone_cache_value(&h.0));
-                    // Object keys can also be prefixes. A classified payload
-                    // directory must not hide their other namespace children.
-                    if found_object_metadata {
-                        continue;
-                    }
                     // Earlier namespace siblings may have completed, but none
                     // can certify coverage of this unresolved parent object.
                     self.coverage_gap |= forward_sweep;

crates/scanner/src/scanner_folder/tests/erasure_probe.rs

@@ -183,117 +183,6 @@ fn charged_ops(root: &std::path::Path) -> u64 {
     status.charged_ops
 }
 
-#[tokio::test]
-#[serial]
-async fn erasure_object_metadata_preserves_child_object_namespaces() {
-    for (corrupt_parent, cached_parent) in [(false, false), (true, false), (false, true)] {
-        let (mut scanner, root) = build_test_scanner().await;
-        let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
-        scanner.old_cache.info.name = "bucket".into();
-        scanner.new_cache.info.name = "bucket".into();
-        scanner.is_erasure_mode = true;
-        bind_budget(&mut scanner, &root, 1_000_000);
-
-        let namespace = format!("parent/{}/nested", Uuid::new_v4());
-        for object in ["parent", "parent/child", namespace.as_str()] {
-            write_test_object_metadata_bytes(&root, "bucket", object, &metadata_for_object_version("bucket", object, None)).await;
-        }
-        let payload = root.join("bucket/parent").join(Uuid::new_v4().to_string());
-        tokio::fs::create_dir_all(&payload)
-            .await
-            .expect("create parent payload directory");
-        tokio::fs::write(payload.join("part.1"), b"shard")
-            .await
-            .expect("create actual payload evidence");
-        if corrupt_parent {
-            tokio::fs::write(root.join("bucket/parent/xl.meta"), b"")
-                .await
-                .expect("corrupt only the parent metadata");
-        }
-        if cached_parent {
-            scanner.old_cache.replace("bucket", "", DataUsageEntry::default());
-            scanner.old_cache.replace(
-                "bucket/parent",
-                "bucket",
-                DataUsageEntry {
-                    objects: 1,
-                    size: 1,
-                    compacted: true,
-                    ..Default::default()
-                },
-            );
-            let interval = data_usage_update_dir_cycles();
-            scanner.old_cache.info.next_cycle = u64::from(
-                (0..interval)
-                    .find(|cycle| hash_path("bucket/parent").mod_(*cycle, interval))
-                    .expect("one normal sampling cycle revisits the old parent-only cache"),
-            );
-        }
-
-        let mut into = DataUsageEntry::default();
-        scanner
-            .scan_folder(
-                CancellationToken::new(),
-                CachedFolder {
-                    name: "bucket".into(),
-                    parent: None,
-                    object_heal_prob_div: 1,
-                },
-                &mut into,
-            )
-            .await
-            .expect("scan object and child namespaces independently");
-        let total = scanner.new_cache.size_recursive("bucket").expect("retain bucket accounting");
-        assert_eq!(total.objects, if corrupt_parent { 2 } else { 3 });
-        assert_eq!(total.size, if corrupt_parent { 2 } else { 3 });
-        assert_eq!(total.failed_objects, usize::from(corrupt_parent));
-        assert_eq!(
-            scanner.new_cache.info.failed_objects.len(),
-            usize::from(corrupt_parent),
-            "payload evidence must not invent missing parent metadata"
-        );
-    }
-}
-
-#[tokio::test]
-#[serial]
-async fn erasure_object_prefix_children_keep_the_object_budget() {
-    let (mut scanner, root) = build_test_scanner().await;
-    let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
-    scanner.old_cache.info.name = "bucket".into();
-    scanner.new_cache.info.name = "bucket".into();
-    scanner.is_erasure_mode = true;
-    bind_budget(&mut scanner, &root, 1_000_000);
-    for object in ["parent", "parent/child", "parent/other"] {
-        write_test_object_metadata(&root, "bucket", object).await;
-    }
-    let visits = metadata_visits::observe(&mut scanner);
-    let parent = CancellationToken::new();
-    let budget = ScannerCycleBudget::new_with_progress_tracking(
-        &parent,
-        crate::scanner_budget::ScannerCycleBudgetConfig {
-            max_objects: Some(2),
-            ..Default::default()
-        },
-    );
-    scanner.budget = budget.clone();
-    let result = scanner
-        .scan_folder(
-            budget.token(),
-            CachedFolder {
-                name: "bucket".into(),
-                parent: None,
-                object_heal_prob_div: 1,
-            },
-            &mut DataUsageEntry::default(),
-        )
-        .await;
-    assert!(result.is_err(), "parent metadata cannot hide children or bypass their budget");
-    assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
-    assert_eq!(budget.progress().0, 2);
-    assert_eq!(visits.load(Ordering::Relaxed), 2);
-}
-
 #[tokio::test]
 #[serial]
 async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
@@ -315,10 +204,7 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                 ctx,
                 folder,
                 &mut DataUsageEntry::default(),
-                FolderScanMode::ErasureCandidate {
-                    reuse_namespace: false,
-                    parent_has_metadata: false,
-                },
+                FolderScanMode::ErasureCandidate { reuse_namespace: false },
                 false,
             )
             .await
@@ -385,10 +271,7 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
             false,
         )
         .await;
@@ -428,10 +311,7 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
             false,
         )
         .await;
@@ -502,16 +382,7 @@ async fn erasure_candidate_defers_metadata_until_part_classification_finishes()
     scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
     let mut into = DataUsageEntry::default();
     let outcome = scanner
-        .scan_folder_inner(
-            ctx,
-            folder,
-            &mut into,
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
-            false,
-        )
+        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false }, false)
         .await
         .expect("classify candidate");
     assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -723,10 +594,7 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
             CancellationToken::new(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
             false,
         )
         .await

crates/utils/src/hash.rs

@@ -735,4 +735,166 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn highwayhash_simd_matches_portable_scalar_and_chunked_slices_across_unaligned_boundaries() {
+        use highway::PortableHash;
+
+        for &size in &[
+            0usize,
+            1,
+            7,
+            15,
+            31,
+            32,
+            33,
+            63,
+            64,
+            65,
+            127,
+            128,
+            129,
+            255,
+            257,
+            1024,
+            4096 + 17,
+            64 * 1024,
+        ] {
+            let payload = generate_compat_test_data(size);
+
+            let mut h_simd = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
+            h_simd.append(&payload);
+            let simd_digest = u8x32_from_u64x4(h_simd.finalize256());
+
+            let mut h_scalar = PortableHash::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
+            h_scalar.append(&payload);
+            let scalar_digest = u8x32_from_u64x4(h_scalar.finalize256());
+
+            assert_eq!(
+                simd_digest, scalar_digest,
+                "HighwayHasher SIMD must match PortableHash scalar byte-for-byte at {size}B"
+            );
+
+            for &chunk_size in &[17usize, 61, 4096] {
+                if size > 4096 && chunk_size < 61 {
+                    continue;
+                }
+                let slices: Vec<&[u8]> = payload.chunks(chunk_size).collect();
+                for algo in [
+                    HashAlgorithm::HighwayHash256S,
+                    HashAlgorithm::HighwayHash256SLegacy,
+                    HashAlgorithm::SHA256,
+                    HashAlgorithm::BLAKE2b512,
+                    HashAlgorithm::Md5,
+                ] {
+                    let contig_out = algo.hash_encode(&payload);
+                    let sliced_out = algo.hash_encode_slices(slices.iter().copied());
+                    assert_eq!(
+                        contig_out.as_ref(),
+                        sliced_out.as_ref(),
+                        "{algo:?} contiguous vs {chunk_size}B sliced digest mismatch at {size}B"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    #[ignore = "utility throughput microbenchmark; run via cargo test -p rustfs-utils --lib --features hash --release -- --ignored --nocapture"]
+    fn microbench_highwayhash_simd_vs_portable_scalar_and_hash_algorithms() {
+        use highway::PortableHash;
+        use std::hint::black_box;
+        use std::time::Instant;
+
+        for &(size, iters) in &[(1024usize, 200usize), (64 * 1024, 40), (1024 * 1024, 8)] {
+            let payload = generate_compat_test_data(size);
+
+            // 1. Compare HighwayHasher (runtime SIMD: AVX2/SSE4.1/NEON) against PortableHash scalar reference.
+            let t_simd = Instant::now();
+            let mut simd_digest = [0u8; 32];
+            for _ in 0..iters {
+                let mut h = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
+                h.append(black_box(&payload));
+                simd_digest = u8x32_from_u64x4(h.finalize256());
+                black_box(&simd_digest);
+            }
+            let simd_elapsed = t_simd.elapsed();
+
+            let t_scalar = Instant::now();
+            let mut scalar_digest = [0u8; 32];
+            for _ in 0..iters {
+                let mut h = PortableHash::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
+                h.append(black_box(&payload));
+                scalar_digest = u8x32_from_u64x4(h.finalize256());
+                black_box(&scalar_digest);
+            }
+            let scalar_elapsed = t_scalar.elapsed();
+
+            assert_eq!(
+                simd_digest, scalar_digest,
+                "HighwayHasher SIMD must match PortableHash scalar byte-for-byte at {size}B"
+            );
+            let total_mib = (size * iters) as f64 / (1024.0 * 1024.0);
+            let simd_mibs = total_mib / simd_elapsed.as_secs_f64().max(1e-9);
+            let scalar_mibs = total_mib / scalar_elapsed.as_secs_f64().max(1e-9);
+            println!(
+                "[microbench][utils::hash] size={size}B HighwayHash256S SIMD={simd_mibs:.1} MiB/s vs PortableScalar={scalar_mibs:.1} MiB/s (speedup={:.2}x)",
+                simd_mibs / scalar_mibs.max(1e-9)
+            );
+
+            // 2. Compare contiguous hash_encode vs 4 KiB chunked hash_encode_slices across all bitrot algorithms.
+            let slices: Vec<&[u8]> = payload.chunks(4096).collect();
+            for algo in [
+                HashAlgorithm::HighwayHash256S,
+                HashAlgorithm::HighwayHash256SLegacy,
+                HashAlgorithm::SHA256,
+                HashAlgorithm::BLAKE2b512,
+                HashAlgorithm::Md5,
+            ] {
+                let t_contig = Instant::now();
+                let mut contig_out = Vec::new();
+                for _ in 0..iters {
+                    contig_out = algo.hash_encode(black_box(&payload)).as_ref().to_vec();
+                    black_box(&contig_out);
+                }
+                let contig_elapsed = t_contig.elapsed();
+
+                let t_sliced = Instant::now();
+                let mut sliced_out = Vec::new();
+                for _ in 0..iters {
+                    sliced_out = algo.hash_encode_slices(black_box(slices.iter().copied())).as_ref().to_vec();
+                    black_box(&sliced_out);
+                }
+                let sliced_elapsed = t_sliced.elapsed();
+
+                assert_eq!(contig_out, sliced_out, "{algo:?} contiguous vs 4KiB sliced digest mismatch at {size}B");
+                let contig_mibs = total_mib / contig_elapsed.as_secs_f64().max(1e-9);
+                let sliced_mibs = total_mib / sliced_elapsed.as_secs_f64().max(1e-9);
+                println!(
+                    "[microbench][utils::hash] size={size}B algo={algo:?} contiguous={contig_mibs:.1} MiB/s vs sliced(4KiB)={sliced_mibs:.1} MiB/s"
+                );
+            }
+        }
+
+        // 3. Key-routing microbenchmark: crc_hash (CRC32) vs sip_hash (SipHash-1-3).
+        let keys: Vec<String> = (0..1024)
+            .map(|i| format!("bucket/prefix-{i}/object-key-{i}.parquet"))
+            .collect();
+        let t_crc = Instant::now();
+        let mut crc_acc = 0usize;
+        for key in &keys {
+            crc_acc ^= crc_hash(black_box(key), 16);
+        }
+        let crc_ns = t_crc.elapsed().as_nanos();
+
+        let t_sip = Instant::now();
+        let mut sip_acc = 0usize;
+        for key in &keys {
+            sip_acc ^= sip_hash(black_box(key), 16, &[0u8; 16]);
+        }
+        black_box(sip_acc);
+        let sip_ns = t_sip.elapsed().as_nanos();
+        assert!(black_box(crc_acc) < 16 && black_box(sip_acc) < 16);
+        println!("[microbench][utils::hash] 1024 object keys: crc_hash={crc_ns}ns total, sip_hash={sip_ns}ns total");
+    }
 }

rustfs/src/app/object/delete.rs

@@ -17,6 +17,7 @@
 use super::*;
 use crate::app::storage_api::object_usecase::DeleteObjectCondition;
 use s3s::dto::ObjectIdentifier;
+use std::borrow::Cow;
 use std::collections::HashSet;
 
 async fn authorize_recursive_delete<T>(
@@ -81,6 +82,19 @@ async fn authorize_recursive_delete<T>(
     result
 }
 
+// XML body keys bypass s3s's URL-path normalization. Match its private slash
+// normalizer without URL-decoding or resolving dot segments in client keys.
+fn normalize_delete_object_key(key: &str) -> Cow<'_, str> {
+    if !key.starts_with('/') {
+        return Cow::Borrowed(key);
+    }
+    let mut normalized = key.split('/').filter(|part| !part.is_empty()).collect::<Vec<_>>().join("/");
+    if key.ends_with('/') {
+        normalized.push('/');
+    }
+    Cow::Owned(normalized)
+}
+
 fn successful_delete_audit_objects(
     delete: &s3s::dto::Delete,
     successful_results: impl IntoIterator<Item = bool>,
@@ -90,7 +104,9 @@ fn successful_delete_audit_objects(
         .iter()
         .zip(successful_results)
         .filter(|(_, successful)| *successful)
-        .map(|(requested, _)| AuditObjectVersion::new(requested.key.clone(), requested.version_id.clone()))
+        .map(|(requested, _)| {
+            AuditObjectVersion::new(normalize_delete_object_key(&requested.key).into_owned(), requested.version_id.clone())
+        })
         .collect()
 }
 
@@ -590,7 +606,10 @@ impl DefaultObjectUsecase {
         // Per-key denials are audited once per denied action, in aggregate, after the loop.
         let mut bulk_denials = crate::storage::access::BulkDenialAudit::start(&mut req)?;
         for (idx, obj_id) in delete.objects.iter().enumerate() {
-            if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
+            // Authorization and every storage side effect must address the same
+            // normalized object; retain the request key only for response entries.
+            let key = normalize_delete_object_key(&obj_id.key);
+            if let Err(err) = validate_object_key(&key, "DeleteObjects") {
                 delete_results[idx].error = Some(s3s::dto::Error {
                     code: Some(err.code().as_str().to_string()),
                     key: Some(obj_id.key.clone()),
@@ -623,7 +642,7 @@ impl DefaultObjectUsecase {
             {
                 let req_info = req_info_mut(&mut req)?;
                 req_info.bucket = Some(bucket.clone());
-                req_info.object = Some(obj_id.key.clone());
+                req_info.object = Some(key.to_string());
                 req_info.version_id = version_id.clone();
             }
 
@@ -670,7 +689,7 @@ impl DefaultObjectUsecase {
                 }
             }
 
-            if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await {
+            if let Err(err) = validate_table_catalog_object_mutation(&bucket, &key).await {
                 delete_results[idx].error = Some(s3s::dto::Error {
                     code: Some("InvalidRequest".to_string()),
                     key: Some(obj_id.key.clone()),
@@ -680,9 +699,9 @@ impl DefaultObjectUsecase {
                 continue;
             }
 
-            let synthetic_version_id = delete_condition.is_none() && version_id.is_none() && is_dir_object(&obj_id.key);
+            let synthetic_version_id = delete_condition.is_none() && version_id.is_none() && is_dir_object(&key);
             let object = ObjectToDelete {
-                object_name: obj_id.key.clone(),
+                object_name: key.into_owned(),
                 version_id: version_uuid,
                 delete_condition,
                 synthetic_version_id,
@@ -910,15 +929,16 @@ impl DefaultObjectUsecase {
         let deleted = (!delete.quiet.unwrap_or_default()).then(|| {
             delete_results
                 .iter()
-                .filter_map(|result| result.delete_object.as_ref().map(|object| (result, object)))
-                .map(|(result, object)| DeletedObject {
+                .enumerate()
+                .filter_map(|(idx, result)| result.delete_object.as_ref().map(|object| (idx, result, object)))
+                .map(|(idx, result, object)| DeletedObject {
                     delete_marker: { if object.delete_marker { Some(true) } else { None } },
                     delete_marker_version_id: delete_marker_response_version_id(
                         object.delete_marker_version_id,
                         object.delete_marker && result.version_suspended && result.requested_version_id.is_none(),
                         result.synthetic_version_id,
                     ),
-                    key: Some(object.object_name.clone()),
+                    key: Some(delete.objects[idx].key.clone()),
                     version_id: delete_entry_response_version_id(
                         object.version_id,
                         object.delete_marker,
@@ -936,7 +956,12 @@ impl DefaultObjectUsecase {
 
         let errors = delete_results
             .iter()
-            .filter_map(|v| v.error.clone())
+            .enumerate()
+            .filter_map(|(idx, result)| {
+                let mut error = result.error.clone()?;
+                error.key = Some(delete.objects[idx].key.clone());
+                Some(error)
+            })
             .collect::<Vec<s3s::dto::Error>>();
         let output = DeleteObjectsOutput {
             deleted,
@@ -2818,11 +2843,57 @@ mod tests {
         assert_eq!(err.message(), Some("Not init"));
     }
 
+    #[test]
+    fn delete_objects_key_normalization_matches_s3s_path_keys() {
+        use s3s::path::{
+            parse_path_style_with_validation_and_normalization, parse_virtual_hosted_style_with_validation_and_normalization,
+        };
+        use s3s::validation::AwsNameValidation;
+
+        for key in [
+            "plain",
+            "/a/b",
+            "//a/b",
+            "/dir///sub//file",
+            "/dir///",
+            "/",
+            "///",
+            "dir//file",
+            "/literal%2F+名",
+            "/dir/../file",
+            "/dir/./file",
+            "/dir\\file",
+            "/a\0b",
+        ] {
+            let normalized = normalize_delete_object_key(key);
+            let path = parse_path_style_with_validation_and_normalization(
+                &format!("/test-bucket/{key}"),
+                &AwsNameValidation::new(),
+                true,
+            )
+            .expect("path-style key should parse before storage validation");
+            let virtual_host = parse_virtual_hosted_style_with_validation_and_normalization(
+                Some("test-bucket"),
+                &format!("/{key}"),
+                &AwsNameValidation::new(),
+                true,
+            )
+            .expect("virtual-hosted key should parse before storage validation");
+            assert_eq!(Some(normalized.as_ref()), path.get_object_key(), "path-style mismatch for {key:?}");
+            assert_eq!(
+                Some(normalized.as_ref()),
+                virtual_host.get_object_key(),
+                "virtual-hosted mismatch for {key:?}"
+            );
+        }
+        assert_eq!(normalize_delete_object_key(""), "");
+    }
+
     #[test]
     fn delete_objects_audit_details_include_only_successful_request_entries() {
         let requested = vec![
             ObjectIdentifier {
-                key: "first-key".to_string(),
+                key: "/first-key".to_string(),
                 version_id: None,
                 ..Default::default()
             },
@@ -2832,7 +2903,7 @@ mod tests {
                 ..Default::default()
             },
             ObjectIdentifier {
-                key: "versioned-key".to_string(),
+                key: "//versioned-key".to_string(),
                 version_id: Some("requested-version".to_string()),
                 ..Default::default()
             },

rustfs/src/table_catalog/tests.rs

@@ -14460,8 +14460,8 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
     assert_ordered(
         method(delete, "fn execute_delete_objects_owned("),
         &[
-            "validate_object_key(&obj_id.key, \"DeleteObjects\")",
-            "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await {",
+            "validate_object_key(&key, \"DeleteObjects\")",
+            "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &key).await {",
             "continue;\n            }",
             "authorized_deletes.push(AuthorizedDelete { idx, object });",
         ],