perf(ecstore,rio,utils): accelerate MD5 ETag hashing and SWAR disk path checks (#9110 )

2026-11-02 01:00:09 +00:00 · 2026-11-02 01:02:26 +08:00
11 changed files with 1009 additions and 90 deletions
@@ -8033,6 +8033,14 @@ impl LocalDisk {
    fn io_get_object_open_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
        #[cfg(target_os = "linux")]
        {
+            if is_clean_relative_segment(bucket) && is_clean_relative_segment(key) {
+                let root = self.io_root();
+                let mut path = PathBuf::with_capacity(root.as_os_str().len() + bucket.len() + key.len() + 2);
+                path.push(root);
+                path.push(bucket);
+                path.push(key);
+                return Ok(path);
+            }
            let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
            check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
            Ok(path)
@@ -8047,6 +8055,11 @@ impl LocalDisk {
    fn io_get_bucket_and_object_open_path(&self, bucket: &str, key: &str) -> Result<(PathBuf, PathBuf)> {
        #[cfg(target_os = "linux")]
        {
+            if is_clean_relative_segment(bucket) && is_clean_relative_segment(key) {
+                let bucket_path = self.io_root().join(bucket);
+                let path = bucket_path.join(key);
+                return Ok((bucket_path, path));
+            }
            let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
            let normalized = check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
            if matches!(normalized, std::borrow::Cow::Owned(_)) {
@@ -12027,8 +12040,7 @@ fn local_disk_object_path(root: &Path, bucket: &str, key: &str) -> Result<PathBu
 #[cfg(not(windows))]
 #[inline]
 fn is_clean_relative_segment(s: &str) -> bool {
-    let b = s.as_bytes();
-    !matches!(b.first(), None | Some(b'/' | b'.')) && !b.ends_with(b"/") && !b.windows(2).any(|w| matches!(w, b"//" | b"/."))
+    !s.starts_with('/') && rustfs_utils::path::unix_is_canonical_ascii_path(s.as_bytes())
 }

 fn build_local_disk_object_path(root: &Path, bucket: &str, key: &str) -> (PathBuf, PathBuf) {
@@ -12075,10 +12087,26 @@ fn build_local_disk_bucket_path(root: &Path, bucket: &str) -> PathBuf {
    bucket_path
 }

+#[inline]
+fn disk_path_starts_with_normalized(path: &Path, base: &Path) -> bool {
+    #[cfg(not(windows))]
+    return rustfs_utils::path::unix_normalized_path_starts_with(path, base);
+    #[cfg(windows)]
+    path.starts_with(base)
+}
+
+#[inline]
+fn disk_path_strip_prefix_normalized<'a>(path: &'a Path, base: &Path) -> Result<&'a Path> {
+    #[cfg(not(windows))]
+    return rustfs_utils::path::unix_normalized_path_strip_prefix(path, base).ok_or(DiskError::InvalidPath);
+    #[cfg(windows)]
+    path.strip_prefix(base).map_err(|_| DiskError::InvalidPath)
+}
+
 fn check_local_disk_valid_object_path(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> {
    let bucket_path = normalize_path_cow(bucket_path);
    let path = normalize_path_cow(path);
-    if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
+    if !disk_path_starts_with_normalized(&bucket_path, root) || !disk_path_starts_with_normalized(&path, &bucket_path) {
        return Err(DiskError::InvalidPath);
    }

@@ -12087,7 +12115,7 @@ fn check_local_disk_valid_object_path(root: &Path, bucket_path: &Path, path: &Pa

 fn check_local_disk_valid_path(root: &Path, path: impl AsRef<Path>) -> Result<()> {
    let path = normalize_path_cow(path.as_ref());
-    if !path.starts_with(root) {
+    if !disk_path_starts_with_normalized(&path, root) {
        return Err(DiskError::InvalidPath);
    }

@@ -12108,7 +12136,7 @@ fn check_local_disk_object_path_components<'a>(
 ) -> Result<std::borrow::Cow<'a, Path>> {
    let bucket_path = normalize_path_cow(bucket_path);
    let path = normalize_path_cow(path);
-    if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
+    if !disk_path_starts_with_normalized(&bucket_path, root) || !disk_path_starts_with_normalized(&path, &bucket_path) {
        return Err(DiskError::InvalidPath);
    }

@@ -12118,7 +12146,7 @@ fn check_local_disk_object_path_components<'a>(
 #[cfg(target_os = "linux")]
 fn check_local_disk_valid_path_at(root: &Path, root_fd: &std::fs::File, path: impl AsRef<Path>) -> Result<()> {
    let path = normalize_path_cow(path.as_ref());
-    if !path.starts_with(root) {
+    if !disk_path_starts_with_normalized(&path, root) {
        return Err(DiskError::InvalidPath);
    }

@@ -12127,7 +12155,7 @@ fn check_local_disk_valid_path_at(root: &Path, root_fd: &std::fs::File, path: im

 #[cfg(target_os = "linux")]
 fn reject_local_disk_symlink_components_at(root: &Path, root_fd: &std::fs::File, path: &Path) -> Result<()> {
-    let relative = path.strip_prefix(root).map_err(|_| DiskError::InvalidPath)?;
+    let relative = disk_path_strip_prefix_normalized(path, root)?;
    match validate_existing_local_disk_prefix_at(root_fd, relative) {
        Ok(()) => Ok(()),
        Err(LocalDiskPathValidationAtError::Unsupported) => reject_local_disk_symlink_components(root, path),
@@ -12155,11 +12183,11 @@ fn validate_existing_local_disk_prefix_at(
        return Ok(());
    }

-    let mut candidate = relative.to_path_buf();
+    let mut candidate = relative;
    loop {
        match openat2(
            root_fd,
-            &candidate,
+            candidate,
            OFlags::PATH | OFlags::CLOEXEC,
            Mode::empty(),
            ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS,
@@ -12168,10 +12196,12 @@ fn validate_existing_local_disk_prefix_at(
            Err(Errno::NOSYS) => return Err(LocalDiskPathValidationAtError::Unsupported),
            Err(Errno::LOOP | Errno::XDEV) => return Err(LocalDiskPathValidationAtError::InvalidPath),
            Err(Errno::NOENT) => {
-                let Some(parent) = candidate.parent().filter(|parent| !parent.as_os_str().is_empty()) else {
+                let Some(parent) =
+                    rustfs_utils::path::unix_path_parent(candidate).filter(|parent| !parent.as_os_str().is_empty())
+                else {
                    return Ok(());
                };
-                candidate = parent.to_path_buf();
+                candidate = parent;
            }
            Err(err) => return Err(LocalDiskPathValidationAtError::Io(err.into())),
        }
@@ -12179,7 +12209,7 @@ fn validate_existing_local_disk_prefix_at(
 }

 fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()> {
-    let relative = path.strip_prefix(root).map_err(|_| DiskError::InvalidPath)?;
+    let relative = disk_path_strip_prefix_normalized(path, root)?;
    let mut current = root.to_path_buf();

    for component in relative.components() {
@@ -12203,12 +12233,11 @@ fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()>
 fn normalize_path_cow(path: &Path) -> std::borrow::Cow<'_, Path> {
    // Windows prefixes retain the existing PathBuf reconstruction semantics.
    #[cfg(not(windows))]
-    if !path.as_os_str().as_encoded_bytes().contains(&b'.')
-        || !path
-            .components()
-            .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir))
    {
-        return std::borrow::Cow::Borrowed(path);
+        let bytes = path.as_os_str().as_encoded_bytes();
+        if bytes != b"." && !bytes.starts_with(b"./") && !rustfs_utils::path::unix_path_has_dotdot_segment(bytes) {
+            return std::borrow::Cow::Borrowed(path);
+        }
    }
    std::borrow::Cow::Owned(normalize_path_components(path))
 }
@@ -4439,12 +4439,18 @@ fn prepare_destination_parent_with_retry(
    base_dir: &Path,
    publication_root: &PublicationRoot,
 ) -> io::Result<ExistingBaseDirectoryGuard> {
-    let prune_budget = parent_prune_budget(parent, base_dir);
+    let mut prune_budget = None;
    let mut attempt = 0;
    loop {
        match mkdir_all_below_existing_base_std(parent, base_dir, publication_root) {
            Ok(preparation) => return Ok(preparation),
-            Err(err) if should_retry_prepare(&err, attempt, prune_budget) => {
+            Err(err)
+                if should_retry_prepare(
+                    &err,
+                    attempt,
+                    *prune_budget.get_or_insert_with(|| parent_prune_budget(parent, base_dir)),
+                ) =>
+            {
                attempt += 1;
            }
            Err(err) => return Err(err),
@@ -4936,13 +4942,9 @@ impl PublicationRoot {
        #[cfg(target_os = "linux")]
        {
            use rustix::fs::{Mode, OFlags};
-            let parent = self.open_directory(
-                path.parent()
-                    .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file parent"))?,
-            )?;
-            let name = path
-                .file_name()
-                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name"))?;
+            let (parent_path, name) = rustfs_utils::path::unix_split_parent_and_file_name(path)
+                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file parent or name"))?;
+            let parent = self.open_directory(parent_path)?;
            let file = open_linux_relative(
                &parent,
                Path::new(name),
@@ -4982,9 +4984,9 @@ impl PublicationRoot {

    #[cfg(target_os = "linux")]
    fn relative_path<'a>(&self, path: &'a Path) -> io::Result<&'a Path> {
-        path.strip_prefix(self.io_path.as_path())
-            .or_else(|_| path.strip_prefix(self.path.as_path()))
-            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root"))
+        rustfs_utils::path::unix_normalized_path_strip_prefix(path, self.io_path.as_path())
+            .or_else(|| rustfs_utils::path::unix_normalized_path_strip_prefix(path, self.path.as_path()))
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root"))
    }

    #[cfg(windows)]
@@ -4996,6 +4998,13 @@ impl PublicationRoot {
    }
 }

+#[cfg(target_os = "linux")]
+#[inline]
+fn linux_relative_path_has_invalid_component(relative: &Path) -> bool {
+    let bytes = relative.as_os_str().as_encoded_bytes();
+    bytes.starts_with(b"/") || rustfs_utils::path::unix_path_has_dotdot_segment(bytes)
+}
+
 /// Open the actual IO handle with the same containment policy as path validation.
 /// Older kernels retain fd-relative traversal; only ENOSYS enables the fallback.
 #[cfg(target_os = "linux")]
@@ -5006,10 +5015,7 @@ fn open_linux_relative(
    mode: rustix::fs::Mode,
 ) -> io::Result<std::fs::File> {
    use rustix::fs::{OFlags, ResolveFlags, openat2};
-    if relative
-        .components()
-        .any(|component| !matches!(component, Component::Normal(_) | Component::CurDir))
-    {
+    if linux_relative_path_has_invalid_component(relative) {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid relative disk path"));
    }
    let relative = if relative.as_os_str().is_empty() {
@@ -5180,11 +5186,8 @@ impl LinuxCommitDirectories {

    fn file_name<'a>(&self, path: &'a Path, source: bool) -> io::Result<&'a std::ffi::OsStr> {
        let parent = if source { &self.source_path } else { &self.destination_path };
-        if path.parent() != Some(parent) {
-            return Err(io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard"));
-        }
-        path.file_name()
-            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name"))
+        rustfs_utils::path::unix_direct_child_file_name(path, parent)
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard"))
    }

    fn validate(&self) -> io::Result<()> {
@@ -5196,8 +5199,8 @@ impl LinuxCommitDirectories {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed"));
        }
        let destination = self.prepare_destination()?;
-        if self.source_path != self.destination_path
-            || (self.source.device, self.source.inode) != (destination.device, destination.inode)
+        if (self.source.device, self.source.inode) != (destination.device, destination.inode)
+            || self.source_path.as_os_str() != self.destination_path.as_os_str()
        {
            let current = self.root.open_directory(&self.destination_path)?.metadata()?;
            if (current.dev(), current.ino()) != (destination.device, destination.inode) {
@@ -5216,7 +5219,10 @@ impl LinuxCommitDirectories {
        loop {
            if let Some(prepared) = prepared {
                let metadata = rustix::fs::statat(&*self.source.file, source_name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW)?;
-                if prepared.path != source || metadata.st_dev != prepared.device || metadata.st_ino != prepared.inode {
+                if (prepared.path.as_os_str() != source.as_os_str() && prepared.path != source)
+                    || metadata.st_dev != prepared.device
+                    || metadata.st_ino != prepared.inode
+                {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        "rename source identity changed while publication was prepared",
@@ -6354,19 +6360,35 @@ fn windows_rename_source_is_allowed(attributes: u32, reparse_tag: u32) -> bool {
 }

 fn existing_base_directory_suffix<'a>(dir_path: &'a Path, base_dir: &Path) -> io::Result<&'a Path> {
-    let relative = dir_path
-        .strip_prefix(base_dir)
-        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory"))?;
-    for component in relative.components() {
-        if !matches!(component, Component::Normal(_) | Component::CurDir) {
+    #[cfg(target_os = "linux")]
+    {
+        let relative = rustfs_utils::path::unix_normalized_path_strip_prefix(dir_path, base_dir).ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory")
+        })?;
+        if linux_relative_path_has_invalid_component(relative) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "rename destination contains an invalid path component",
            ));
        }
+        Ok(relative)
    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        let relative = dir_path.strip_prefix(base_dir).map_err(|_| {
+            io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory")
+        })?;
+        for component in relative.components() {
+            if !matches!(component, Component::Normal(_) | Component::CurDir) {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    "rename destination contains an invalid path component",
+                ));
+            }
+        }

-    Ok(relative)
+        Ok(relative)
+    }
 }

 /// Missing state never downgrades surviving native control records to legacy.
@@ -429,6 +429,7 @@ impl WritePlan {
        let checksum = reader.content_hash().clone();
        let trailer = reader.get_trailer().cloned();

+        let delegate_inner_etag = self.compression.is_none() && reader.is_etag_reader();
        let encrypted = self.encryption.is_some();
        if let Some(algorithm) = self.compression {
            reader = HashReader::from_reader(
@@ -448,7 +449,8 @@ impl WritePlan {
                }
                WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
            };
-            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
+            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?
+                .with_inner_etag_delegation(delegate_inner_etag);
            reader = match encryption.mode {
                WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
@@ -2432,7 +2432,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            let multipart_part_size = data.size();
            let stream = mem::replace(
                &mut data.stream,
-                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
+                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, true)?,
            );

            let write_path = classify_multipart_part_write_path(multipart_part_size, fi.erasure.block_size);
@@ -4534,7 +4534,7 @@ impl SetDisks {

            let stream = mem::replace(
                &mut data.stream,
-                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
+                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, true)?,
            );

            let small_size_hint = if matches!(write_path, SmallWritePath::Inline | SmallWritePath::SingleBlockNonInline) {
@@ -1081,7 +1081,7 @@ impl ChecksumHasher for Sha512Hasher {
 /// MD5 hasher for the ADDITIONAL checksum (x-amz-checksum-md5). Separate from the
 /// legacy Content-MD5 / ETag machinery — this only serves the flexible-checksum path.
 pub struct Md5Hasher {
-    hasher: md5::Md5,
+    hasher: rustfs_utils::FastMd5,
 }

 impl Default for Md5Hasher {
@@ -1092,14 +1092,14 @@ impl Default for Md5Hasher {

 impl Md5Hasher {
    pub fn new() -> Self {
-        use md5::Digest as _;
-        Self { hasher: md5::Md5::new() }
+        Self {
+            hasher: rustfs_utils::FastMd5::new(),
+        }
    }
 }

 impl Write for Md5Hasher {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        use md5::Digest as _;
        self.hasher.update(buf);
        Ok(buf.len())
    }
@@ -1111,13 +1111,11 @@ impl Write for Md5Hasher {

 impl ChecksumHasher for Md5Hasher {
    fn finalize(&mut self) -> Vec<u8> {
-        use md5::Digest as _;
-        self.hasher.clone().finalize().to_vec()
+        self.hasher.finalize_bytes().to_vec()
    }

    fn reset(&mut self) {
-        use md5::Digest as _;
-        self.hasher = md5::Md5::new();
+        self.hasher.reset();
    }
 }

@@ -15,8 +15,8 @@
 use crate::compress_index::{Index, TryGetIndex};
 use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
 use hmac::{Hmac, KeyInit, Mac};
-use md5::{Digest, Md5};
 use pin_project_lite::pin_project;
+use rustfs_utils::FastMd5;
 use sha2::Sha256;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -27,11 +27,12 @@ pin_project! {
    pub struct  EtagReader<R> {
        #[pin]
        pub inner: R,
-        pub md5: Md5,
+        pub md5: FastMd5,
        pub finished: bool,
        pub checksum: Option<String>,
        resolved_etag: Option<String>,
        encrypted_etag_mac: Option<Hmac<Sha256>>,
+        delegate_inner_etag: bool,
    }
 }

@@ -39,11 +40,12 @@ impl<R> EtagReader<R> {
    pub fn new(inner: R, checksum: Option<String>) -> Self {
        Self {
            inner,
-            md5: Md5::new(),
+            md5: FastMd5::new(),
            finished: false,
            checksum,
            resolved_etag: None,
            encrypted_etag_mac: None,
+            delegate_inner_etag: false,
        }
    }

@@ -58,6 +60,17 @@ impl<R> EtagReader<R> {
        Ok(reader)
    }

+    /// Reuse the inner reader's plaintext ETag when deriving an encrypted ETag,
+    /// avoiding a redundant second MD5 pass over the same plaintext stream.
+    pub fn with_inner_etag_delegation(mut self, delegate: bool) -> Self
+    where
+        R: EtagResolvable,
+    {
+        self.delegate_inner_etag =
+            delegate && self.encrypted_etag_mac.is_some() && self.checksum.is_none() && self.inner.is_etag_reader();
+        self
+    }
+
    /// Get the final md5 value (etag) as a hex string, only compute once.
    /// Can be called multiple times, always returns the same result after finished.
    pub fn get_etag(&mut self) -> String {
@@ -65,8 +78,8 @@ impl<R> EtagReader<R> {
            return etag.clone();
        }

-        let etag = self.md5.clone().finalize().to_vec();
-        let etag = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower);
+        let digest = self.md5.finalize_bytes();
+        let etag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
        self.resolved_etag = Some(etag.clone());
        etag
    }
@@ -87,30 +100,32 @@ where
        if let Poll::Ready(Ok(())) = &poll {
            let filled = &buf.filled()[orig_filled..];
            if !filled.is_empty() {
-                this.md5.update(filled);
+                if !*this.delegate_inner_etag {
+                    this.md5.update(filled);
+                }
            } else {
                // EOF
                *this.finished = true;
-                let etag = if let Some(etag) = this.resolved_etag.as_ref() {
-                    etag.clone()
-                } else {
-                    let etag = this.md5.clone().finalize().to_vec();
-                    let etag = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower);
-                    *this.resolved_etag = Some(etag.clone());
-                    etag
-                };
+                if let Some(checksum) = this.checksum {
+                    let etag = if let Some(etag) = this.resolved_etag.as_ref() {
+                        etag.clone()
+                    } else {
+                        let digest = this.md5.finalize_bytes();
+                        let etag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
+                        *this.resolved_etag = Some(etag.clone());
+                        etag
+                    };

-                if let Some(checksum) = this.checksum
-                    && *checksum != etag
-                {
-                    error!("Checksum mismatch, expected={:?}, actual={:?}", checksum, etag);
-                    return Poll::Ready(Err(std::io::Error::new(
-                        std::io::ErrorKind::InvalidData,
-                        BadDigest {
-                            expected_md5: checksum.clone(),
-                            calculated_md5: etag,
-                        },
-                    )));
+                    if *checksum != etag {
+                        error!("Checksum mismatch, expected={:?}, actual={:?}", checksum, etag);
+                        return Poll::Ready(Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidData,
+                            BadDigest {
+                                expected_md5: checksum.clone(),
+                                calculated_md5: etag,
+                            },
+                        )));
+                    }
                }
            }
        }
@@ -118,16 +133,24 @@ where
    }
 }

-impl<R> EtagResolvable for EtagReader<R> {
+impl<R> EtagResolvable for EtagReader<R>
+where
+    R: EtagResolvable,
+{
    fn is_etag_reader(&self) -> bool {
-        true
+        !self.delegate_inner_etag || self.inner.is_etag_reader()
    }
    fn try_resolve_etag(&mut self) -> Option<String> {
        if let Some(mut mac) = self.encrypted_etag_mac.clone() {
            if !self.finished {
                return None;
            }
-            mac.update(self.get_etag().as_bytes());
+            let plaintext_etag = if self.delegate_inner_etag {
+                self.inner.try_resolve_etag()?
+            } else {
+                self.get_etag()
+            };
+            mac.update(plaintext_etag.as_bytes());
            let protected = mac.finalize().into_bytes();
            return Some(hex_simd::encode_to_string(&protected[..16], hex_simd::AsciiCase::Lower));
        }
@@ -167,6 +190,7 @@ where
 #[cfg(test)]
 mod tests {
    use super::*;
+    use md5::{Digest, Md5};
    use rand::RngExt;
    use std::io::Cursor;
    use tokio::io::{AsyncReadExt, BufReader};
@@ -340,4 +364,54 @@ mod tests {
        assert_eq!(digest.expected_md5, wrong_checksum);
        assert_eq!(digest.calculated_md5, calculated_md5);
    }
+
+    #[tokio::test]
+    async fn encrypted_etag_reader_with_inner_delegation_matches_standalone() {
+        let data = b"delegated encrypted etag payload across multiple 64-byte MD5 blocks for verification";
+        let key = [0x42; 32];
+        let nonce = [0x24; 12];
+
+        let mut standalone =
+            EtagReader::new_encrypted(Cursor::new(data.as_slice()), key, nonce).expect("standalone encrypted EtagReader");
+        let mut out_standalone = Vec::new();
+        standalone.read_to_end(&mut out_standalone).await.expect("read standalone");
+        let expected_tag = standalone.try_resolve_etag().expect("standalone etag");
+
+        let inner =
+            crate::HashReader::from_stream(Cursor::new(data.as_slice()), data.len() as i64, data.len() as i64, None, None, false)
+                .expect("inner HashReader");
+        assert!(inner.is_etag_reader(), "HashReader with diskable_md5=false must report is_etag_reader");
+        let mut delegated = EtagReader::new_encrypted(inner, key, nonce)
+            .expect("delegated encrypted EtagReader")
+            .with_inner_etag_delegation(true);
+        let mut out_delegated = Vec::new();
+        delegated.read_to_end(&mut out_delegated).await.expect("read delegated");
+        assert_eq!(out_delegated, data);
+        assert_eq!(
+            delegated.try_resolve_etag().as_deref(),
+            Some(expected_tag.as_str()),
+            "delegated encrypted EtagReader must produce identical HMAC ETag without re-hashing plaintext"
+        );
+
+        let truncated = crate::LimitReader::new(
+            crate::HashReader::from_stream(Cursor::new(data.as_slice()), data.len() as i64, data.len() as i64, None, None, false)
+                .expect("inner HashReader for limit"),
+            16,
+        );
+        assert!(
+            !truncated.is_etag_reader(),
+            "truncating LimitReader must not advertise inner ETag delegation"
+        );
+
+        let mut non_etag_inner = EtagReader::new_encrypted(Cursor::new(data.as_slice()), key, nonce)
+            .expect("encrypted EtagReader over Cursor")
+            .with_inner_etag_delegation(true);
+        let mut out_cursor = Vec::new();
+        non_etag_inner.read_to_end(&mut out_cursor).await.expect("read cursor");
+        assert_eq!(
+            non_etag_inner.try_resolve_etag().as_deref(),
+            Some(expected_tag.as_str()),
+            "delegation request on non-EtagResolvable inner must keep outer MD5 hashing active"
+        );
+    }
 }
@@ -630,6 +630,10 @@ impl AsyncRead for HashReader {
 }

 impl EtagResolvable for HashReader {
+    fn is_etag_reader(&self) -> bool {
+        !self.diskable_md5 && self.inner.is_etag_reader()
+    }
+
    fn try_resolve_etag(&mut self) -> Option<String> {
        if self.diskable_md5 {
            return None;
@@ -201,10 +201,31 @@ where
    boxed_reader(WarpReader::new(reader))
 }

+impl<T> EtagResolvable for std::io::Cursor<T> {}
+
+impl EtagResolvable for &[u8] {}
+
+impl<R> EtagResolvable for tokio::io::BufReader<R>
+where
+    R: tokio::io::AsyncRead + EtagResolvable,
+{
+    fn is_etag_reader(&self) -> bool {
+        self.get_ref().is_etag_reader()
+    }
+
+    fn try_resolve_etag(&mut self) -> Option<String> {
+        self.get_mut().try_resolve_etag()
+    }
+}
+
 impl<T> EtagResolvable for Box<T>
 where
    T: EtagResolvable + ?Sized,
 {
+    fn is_etag_reader(&self) -> bool {
+        self.as_ref().is_etag_reader()
+    }
+
    fn try_resolve_etag(&mut self) -> Option<String> {
        self.as_mut().try_resolve_etag()
    }
@@ -14,9 +14,300 @@

 use blake2::Blake2b512;
 use highway::{HighwayHash, HighwayHasher, Key};
-use md5::Md5;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
+use std::hint::black_box;
+
+const MD5_INITIAL_STATE: [u32; 4] = [0x6745_2301, 0xefcd_ab89, 0x98ba_dcfe, 0x1032_5476];
+
+const MD5_RC: [u32; 64] = [
+    0xd76a_a478,
+    0xe8c7_b756,
+    0x2420_70db,
+    0xc1bd_ceee,
+    0xf57c_0faf,
+    0x4787_c62a,
+    0xa830_4613,
+    0xfd46_9501,
+    0x6980_98d8,
+    0x8b44_f7af,
+    0xffff_5bb1,
+    0x895c_d7be,
+    0x6b90_1122,
+    0xfd98_7193,
+    0xa679_438e,
+    0x49b4_0821,
+    0xf61e_2562,
+    0xc040_b340,
+    0x265e_5a51,
+    0xe9b6_c7aa,
+    0xd62f_105d,
+    0x0244_1453,
+    0xd8a1_e681,
+    0xe7d3_fbc8,
+    0x21e1_cde6,
+    0xc337_07d6,
+    0xf4d5_0d87,
+    0x455a_14ed,
+    0xa9e3_e905,
+    0xfcef_a3f8,
+    0x676f_02d9,
+    0x8d2a_4c8a,
+    0xfffa_3942,
+    0x8771_f681,
+    0x6d9d_6122,
+    0xfde5_380c,
+    0xa4be_ea44,
+    0x4bde_cfa9,
+    0xf6bb_4b60,
+    0xbebf_bc70,
+    0x289b_7ec6,
+    0xeaa1_27fa,
+    0xd4ef_3085,
+    0x0488_1d05,
+    0xd9d4_d039,
+    0xe6db_99e5,
+    0x1fa2_7cf8,
+    0xc4ac_5665,
+    0xf429_2244,
+    0x432a_ff97,
+    0xab94_23a7,
+    0xfc93_a039,
+    0x655b_59c3,
+    0x8f0c_cc92,
+    0xffef_f47d,
+    0x8584_5dd1,
+    0x6fa8_7e4f,
+    0xfe2c_e6e0,
+    0xa301_4314,
+    0x4e08_11a1,
+    0xf753_7e82,
+    0xbd3a_f235,
+    0x2ad7_d2bb,
+    0xeb86_d391,
+];
+
+#[inline(always)]
+fn md5_op_f(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    // Keep `w + m + c` off the `x` critical dependency path from the previous round.
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    let f = z ^ (x & (y ^ z));
+    base.wrapping_add(f).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_g(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c)).wrapping_add(y & !z);
+    base.wrapping_add(x & z).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_h(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    base.wrapping_add(x ^ (y ^ z)).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_i(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    base.wrapping_add(y ^ (x | !z)).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_word(words: &[[u8; 4]], idx: usize) -> u32 {
+    u32::from_le_bytes(words[idx])
+}
+
+#[inline]
+pub fn compress_md5_blocks(state: &mut [u32; 4], blocks: &[[u8; 64]]) {
+    let [mut a, mut b, mut c, mut d] = *state;
+    for block in blocks {
+        let words = block.as_chunks::<4>().0;
+        let aa = a;
+        let bb = b;
+        let cc = c;
+        let dd = d;
+
+        a = md5_op_f(a, b, c, d, md5_word(words, 0), MD5_RC[0], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 1), MD5_RC[1], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 2), MD5_RC[2], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 3), MD5_RC[3], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 4), MD5_RC[4], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 5), MD5_RC[5], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 6), MD5_RC[6], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 7), MD5_RC[7], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 8), MD5_RC[8], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 9), MD5_RC[9], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 10), MD5_RC[10], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 11), MD5_RC[11], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 12), MD5_RC[12], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 13), MD5_RC[13], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 14), MD5_RC[14], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 15), MD5_RC[15], 22);
+
+        a = md5_op_g(a, b, c, d, md5_word(words, 1), MD5_RC[16], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 6), MD5_RC[17], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 11), MD5_RC[18], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 0), MD5_RC[19], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 5), MD5_RC[20], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 10), MD5_RC[21], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 15), MD5_RC[22], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 4), MD5_RC[23], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 9), MD5_RC[24], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 14), MD5_RC[25], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 3), MD5_RC[26], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 8), MD5_RC[27], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 13), MD5_RC[28], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 2), MD5_RC[29], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 7), MD5_RC[30], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 12), MD5_RC[31], 20);
+
+        a = md5_op_h(a, b, c, d, md5_word(words, 5), MD5_RC[32], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 8), MD5_RC[33], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 11), MD5_RC[34], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 14), MD5_RC[35], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 1), MD5_RC[36], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 4), MD5_RC[37], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 7), MD5_RC[38], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 10), MD5_RC[39], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 13), MD5_RC[40], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 0), MD5_RC[41], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 3), MD5_RC[42], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 6), MD5_RC[43], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 9), MD5_RC[44], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 12), MD5_RC[45], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 15), MD5_RC[46], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 2), MD5_RC[47], 23);
+
+        a = md5_op_i(a, b, c, d, md5_word(words, 0), MD5_RC[48], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 7), MD5_RC[49], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 14), MD5_RC[50], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 5), MD5_RC[51], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 12), MD5_RC[52], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 3), MD5_RC[53], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 10), MD5_RC[54], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 1), MD5_RC[55], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 8), MD5_RC[56], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 15), MD5_RC[57], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 6), MD5_RC[58], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 13), MD5_RC[59], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 4), MD5_RC[60], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 11), MD5_RC[61], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 2), MD5_RC[62], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 9), MD5_RC[63], 21);
+
+        a = a.wrapping_add(aa);
+        b = b.wrapping_add(bb);
+        c = c.wrapping_add(cc);
+        d = d.wrapping_add(dd);
+    }
+    *state = [a, b, c, d];
+}
+
+#[inline]
+fn finalize_md5_state(mut state: [u32; 4], tail: &[u8], total_bytes: u64) -> [u8; 16] {
+    debug_assert!(tail.len() < 64);
+    let mut pad = [0u8; 128];
+    pad[..tail.len()].copy_from_slice(tail);
+    pad[tail.len()] = 0x80;
+    let bit_len = total_bytes.wrapping_mul(8).to_le_bytes();
+    if tail.len() < 56 {
+        pad[56..64].copy_from_slice(&bit_len);
+        compress_md5_blocks(&mut state, pad[..64].as_chunks::<64>().0);
+    } else {
+        pad[120..128].copy_from_slice(&bit_len);
+        compress_md5_blocks(&mut state, pad.as_chunks::<64>().0);
+    }
+    let mut out = [0u8; 16];
+    out[0..4].copy_from_slice(&state[0].to_le_bytes());
+    out[4..8].copy_from_slice(&state[1].to_le_bytes());
+    out[8..12].copy_from_slice(&state[2].to_le_bytes());
+    out[12..16].copy_from_slice(&state[3].to_le_bytes());
+    out
+}
+
+/// Streaming MD5 hasher using critical-path ILP reassociation and direct slice chunking.
+#[derive(Clone, Debug)]
+pub struct FastMd5 {
+    state: [u32; 4],
+    buf: [u8; 64],
+    buf_len: usize,
+    total_bytes: u64,
+}
+
+impl Default for FastMd5 {
+    #[inline]
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl FastMd5 {
+    #[inline]
+    pub const fn new() -> Self {
+        Self {
+            state: MD5_INITIAL_STATE,
+            buf: [0u8; 64],
+            buf_len: 0,
+            total_bytes: 0,
+        }
+    }
+
+    #[inline]
+    pub fn reset(&mut self) {
+        self.state = MD5_INITIAL_STATE;
+        self.buf_len = 0;
+        self.total_bytes = 0;
+    }
+
+    #[inline]
+    pub fn update(&mut self, mut data: &[u8]) {
+        if data.is_empty() {
+            return;
+        }
+        self.total_bytes = self.total_bytes.wrapping_add(data.len() as u64);
+        if self.buf_len > 0 {
+            let need = 64 - self.buf_len;
+            if data.len() < need {
+                self.buf[self.buf_len..self.buf_len + data.len()].copy_from_slice(data);
+                self.buf_len += data.len();
+                return;
+            }
+            self.buf[self.buf_len..64].copy_from_slice(&data[..need]);
+            compress_md5_blocks(&mut self.state, std::slice::from_ref(&self.buf));
+            self.buf_len = 0;
+            data = &data[need..];
+        }
+        let (blocks, remainder) = data.as_chunks::<64>();
+        if !blocks.is_empty() {
+            compress_md5_blocks(&mut self.state, blocks);
+        }
+        if !remainder.is_empty() {
+            self.buf[..remainder.len()].copy_from_slice(remainder);
+            self.buf_len = remainder.len();
+        }
+    }
+
+    #[inline]
+    pub fn finalize_bytes(&self) -> [u8; 16] {
+        finalize_md5_state(self.state, &self.buf[..self.buf_len], self.total_bytes)
+    }
+
+    #[inline]
+    pub fn finalize(self) -> [u8; 16] {
+        self.finalize_bytes()
+    }
+
+    #[inline]
+    pub fn digest(data: &[u8]) -> [u8; 16] {
+        let mut state = MD5_INITIAL_STATE;
+        let (blocks, remainder) = data.as_chunks::<64>();
+        if !blocks.is_empty() {
+            compress_md5_blocks(&mut state, blocks);
+        }
+        finalize_md5_state(state, remainder, data.len() as u64)
+    }
+}

 /// Magic HH-256 key: HH-256 hash of first 100 decimals of π as utf-8 with zero key.
 const MAGIC_HIGHWAY_HASH256_KEY: [u8; 32] = [
@@ -113,7 +404,7 @@ impl HashAlgorithm {
    #[inline]
    pub fn hash_encode(&self, data: &[u8]) -> impl AsRef<[u8]> {
        match self {
-            HashAlgorithm::Md5 => HashEncoded::Md5(Md5::digest(data).into()),
+            HashAlgorithm::Md5 => HashEncoded::Md5(FastMd5::digest(data)),
            HashAlgorithm::HighwayHash256 => {
                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
                hasher.append(data);
@@ -148,11 +439,11 @@ impl HashAlgorithm {
    {
        match self {
            HashAlgorithm::Md5 => {
-                let mut hasher = Md5::new();
+                let mut hasher = FastMd5::new();
                for slice in slices {
                    hasher.update(slice);
                }
-                HashEncoded::Md5(hasher.finalize().into())
+                HashEncoded::Md5(hasher.finalize_bytes())
            }
            HashAlgorithm::HighwayHash256 => {
                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
@@ -424,4 +715,24 @@ mod tests {
        let blake_hash2 = HashAlgorithm::BLAKE2b512.hash_encode(data2);
        assert_ne!(blake_hash1.as_ref(), blake_hash2.as_ref());
    }
+
+    #[test]
+    fn fast_md5_matches_reference_across_boundaries_and_chunkings() {
+        use md5::{Digest as _, Md5};
+
+        for len in (0..=260).chain([511, 512, 513, 1024, 4096, 8193]) {
+            let data = generate_compat_test_data(len);
+            let expected: [u8; 16] = Md5::digest(&data).into();
+            assert_eq!(FastMd5::digest(&data), expected, "FastMd5::digest mismatch at len={len}");
+
+            for step in [1usize, 3, 7, 16, 31, 55, 56, 63, 64, 65, 128] {
+                let mut fast = FastMd5::new();
+                for chunk in data.chunks(step) {
+                    fast.update(chunk);
+                }
+                assert_eq!(fast.finalize_bytes(), expected, "FastMd5 streaming mismatch at len={len}, step={step}");
+                assert_eq!(fast.finalize(), expected, "FastMd5 repeated finalize mismatch at len={len}, step={step}");
+            }
+        }
+    }
 }
@@ -694,6 +694,358 @@ pub fn trim_etag(etag: &str) -> String {
    etag.trim_matches('"').to_string()
 }

+/// Exact 64-bit SWAR byte equality mask for `b'.'` (`0x2e`) with zero cross-byte carry.
+/// Bit `7 + 8 * k` is set in the returned mask iff byte `k` of `w` equals `b'.'`.
+#[inline(always)]
+const fn swar_exact_dot_mask(w: u64) -> u64 {
+    let x = w ^ 0x2e2e_2e2e_2e2e_2e2e;
+    let low = (x & 0x7f7f_7f7f_7f7f_7f7f).wrapping_add(0x7f7f_7f7f_7f7f_7f7f);
+    !(low | x) & 0x8080_8080_8080_8080
+}
+
+/// Exact 64-bit SWAR mask for `b'.'` (`0x2e`) or `b'/'` (`0x2f`) across all byte values `0x00..=0xFF`.
+/// Since `0x2e` and `0x2f` differ only in bit 0, `(b & 0xfe) == 0x2e` holds iff `b` is `.` or `/`.
+#[inline(always)]
+const fn swar_exact_dot_or_slash_mask(w: u64) -> u64 {
+    swar_exact_dot_mask(w & 0xfefe_fefe_fefe_fefe)
+}
+
+#[inline(always)]
+const fn is_unix_dot_or_dotdot_at(path: &[u8], dot_idx: usize) -> bool {
+    if dot_idx > 0 && path[dot_idx - 1] != b'/' {
+        return false;
+    }
+    let next = dot_idx + 1;
+    if next == path.len() || path[next] == b'/' {
+        return true;
+    }
+    path[next] == b'.' && (next + 1 == path.len() || path[next + 1] == b'/')
+}
+
+#[inline(always)]
+const fn is_unix_dotdot_at(path: &[u8], dot_idx: usize) -> bool {
+    if dot_idx > 0 && path[dot_idx - 1] != b'/' {
+        return false;
+    }
+    let next = dot_idx + 1;
+    next < path.len() && path[next] == b'.' && (next + 1 == path.len() || path[next + 1] == b'/')
+}
+
+/// Returns `true` iff `path` (interpreted with Unix `/` separators) contains a `..` segment.
+/// Scans 8 bytes at a time using carry-free 64-bit SWAR so ordinary filenames containing dots
+/// (such as `xl.meta`, `part.1`, or `.rustfs.sys`) do not trigger component parsing.
+#[inline]
+pub const fn unix_path_has_dotdot_segment(path: &[u8]) -> bool {
+    let (chunks, rem) = path.as_chunks::<8>();
+    let mut c_idx = 0;
+    while c_idx < chunks.len() {
+        let mut mask = swar_exact_dot_mask(u64::from_le_bytes(chunks[c_idx]));
+        while mask != 0 {
+            let lane = (mask.trailing_zeros() >> 3) as usize;
+            if is_unix_dotdot_at(path, c_idx * 8 + lane) {
+                return true;
+            }
+            mask &= mask - 1;
+        }
+        c_idx += 1;
+    }
+    let base = chunks.len() * 8;
+    let mut r_idx = 0;
+    while r_idx < rem.len() {
+        if rem[r_idx] == b'.' && is_unix_dotdot_at(path, base + r_idx) {
+            return true;
+        }
+        r_idx += 1;
+    }
+    false
+}
+
+#[inline(always)]
+const fn is_invalid_adjacent_dot_slash_pair(path: &[u8], idx: usize) -> bool {
+    let prev = path[idx - 1];
+    let curr = path[idx];
+    if prev == b'/' && curr == b'/' {
+        return true;
+    }
+    if prev == b'/' && curr == b'.' {
+        return is_unix_dot_or_dotdot_at(path, idx);
+    }
+    if prev == b'.' && curr == b'/' {
+        return is_unix_dot_or_dotdot_at(path, idx - 1);
+    }
+    if prev == b'.' && curr == b'.' {
+        return is_unix_dotdot_at(path, idx - 1);
+    }
+    false
+}
+
+/// Returns `true` when `path` is a non-empty Unix path with no repeated slashes (`//`),
+/// no trailing slash (except `"/"`), and no `.` or `..` segments.
+#[inline]
+pub const fn unix_is_canonical_ascii_path(path: &[u8]) -> bool {
+    if path.is_empty() {
+        return false;
+    }
+    if path.len() == 1 {
+        return path[0] != b'.';
+    }
+    let first = path[0];
+    let last = path[path.len() - 1];
+    if last == b'/' || (first == b'.' && is_unix_dot_or_dotdot_at(path, 0)) {
+        return false;
+    }
+    let (chunks, rem) = path.as_chunks::<8>();
+    let mut c_idx = 0;
+    let mut prev_hi = 0u64;
+    while c_idx < chunks.len() {
+        let w = u64::from_le_bytes(chunks[c_idx]);
+        let m = swar_exact_dot_or_slash_mask(w);
+        let mut pair_mask = m & ((m << 8) | prev_hi);
+        while pair_mask != 0 {
+            let lane = (pair_mask.trailing_zeros() >> 3) as usize;
+            let idx = c_idx * 8 + lane;
+            if idx > 0 && is_invalid_adjacent_dot_slash_pair(path, idx) {
+                return false;
+            }
+            pair_mask &= pair_mask - 1;
+        }
+        prev_hi = m >> 56;
+        c_idx += 1;
+    }
+    let base = chunks.len() * 8;
+    let mut r_idx = 0;
+    while r_idx < rem.len() {
+        let b = rem[r_idx];
+        let idx = base + r_idx;
+        if (b == b'/' || b == b'.') && idx > 0 {
+            let prev = path[idx - 1];
+            if (prev == b'/' || prev == b'.') && is_invalid_adjacent_dot_slash_pair(path, idx) {
+                return false;
+            }
+        }
+        r_idx += 1;
+    }
+    true
+}
+
+#[inline(always)]
+const fn trim_trailing_slashes_and_curdir(mut bytes: &[u8]) -> &[u8] {
+    while bytes.len() > 1 {
+        if bytes[bytes.len() - 1] == b'/' {
+            bytes = bytes.split_at(bytes.len() - 1).0;
+        } else if bytes.len() >= 2 && bytes[bytes.len() - 2] == b'/' && bytes[bytes.len() - 1] == b'.' {
+            if bytes.len() == 2 {
+                bytes = b"/";
+                break;
+            }
+            bytes = bytes.split_at(bytes.len() - 2).0;
+        } else {
+            break;
+        }
+    }
+    bytes
+}
+
+#[inline(always)]
+const fn strip_leading_slashes_and_curdir(mut rem: &[u8]) -> &[u8] {
+    while !rem.is_empty() {
+        if rem[0] == b'/' {
+            rem = rem.split_at(1).1;
+        } else if rem.len() == 1 && rem[0] == b'.' {
+            rem = b"";
+        } else if rem.len() >= 2 && rem[0] == b'.' && rem[1] == b'/' {
+            rem = rem.split_at(2).1;
+        } else {
+            break;
+        }
+    }
+    rem
+}
+
+#[inline(always)]
+fn unix_bytes_to_os_str(bytes: &[u8]) -> Option<&std::ffi::OsStr> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::ffi::OsStrExt;
+        Some(std::ffi::OsStr::from_bytes(bytes))
+    }
+    #[cfg(not(unix))]
+    {
+        std::str::from_utf8(bytes).ok().map(std::ffi::OsStr::new)
+    }
+}
+
+#[inline(always)]
+fn unix_bytes_to_path(bytes: &[u8]) -> Option<&Path> {
+    unix_bytes_to_os_str(bytes).map(Path::new)
+}
+
+/// Fast Unix `Path::starts_with` when both `path` and `base` have already been normalized
+/// (i.e. contain no `..` segments or leading `.`). Performs a single slice prefix comparison.
+#[inline]
+pub fn unix_normalized_path_starts_with(path: &Path, base: &Path) -> bool {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(base.as_os_str().as_encoded_bytes());
+    if b.is_empty() {
+        return true;
+    }
+    if b == b"/" {
+        return p.starts_with(b"/");
+    }
+    if p.starts_with(b) {
+        return p.len() == b.len() || p[b.len()] == b'/';
+    }
+    if b.contains(&b'.') || b.windows(2).any(|w| w == b"//") {
+        return path.starts_with(base);
+    }
+    false
+}
+
+/// Fast Unix `Path::strip_prefix` when both `path` and `base` have already been normalized
+/// (i.e. contain no `..` segments or leading `.`). Performs a single slice prefix comparison.
+#[inline]
+pub fn unix_normalized_path_strip_prefix<'a>(path: &'a Path, base: &Path) -> Option<&'a Path> {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(base.as_os_str().as_encoded_bytes());
+    if b.is_empty() {
+        return Some(path);
+    }
+    if b == b"/" {
+        if !p.starts_with(b"/") {
+            return None;
+        }
+        let mut rem = strip_leading_slashes_and_curdir(&p[1..]);
+        if rem.ends_with(b"/") || rem.ends_with(b"/.") {
+            rem = trim_trailing_slashes_and_curdir(rem);
+            if rem == b"/" {
+                rem = b"";
+            }
+        }
+        if let Some(res) = unix_bytes_to_path(rem) {
+            return Some(res);
+        }
+    } else if p.starts_with(b) {
+        if p.len() == b.len() {
+            return Some(Path::new(""));
+        }
+        if p[b.len()] == b'/' {
+            let mut rem = strip_leading_slashes_and_curdir(&p[b.len() + 1..]);
+            if rem.ends_with(b"/") || rem.ends_with(b"/.") {
+                rem = trim_trailing_slashes_and_curdir(rem);
+                if rem == b"/" {
+                    rem = b"";
+                }
+            }
+            if let Some(res) = unix_bytes_to_path(rem) {
+                return Some(res);
+            }
+        }
+        return None;
+    }
+    if b.contains(&b'.') || b.windows(2).any(|w| w == b"//") {
+        return path.strip_prefix(base).ok();
+    }
+    None
+}
+
+/// Fast Unix `Path::parent` that slices at the last `/` via SIMD `memrchr` without scanning
+/// earlier path components.
+#[inline]
+pub fn unix_path_parent(path: &Path) -> Option<&Path> {
+    let p = path.as_os_str().as_encoded_bytes();
+    if p.is_empty() || p == b"/" {
+        return None;
+    }
+    if let Some(idx) = p.iter().rposition(|&b| b == b'/') {
+        if idx + 1 < p.len() {
+            let tail = &p[idx + 1..];
+            if tail != b"." && tail != b".." {
+                if idx == 0 {
+                    return Some(Path::new("/"));
+                }
+                if !(idx == 1 && p[0] == b'.') {
+                    let parent_bytes = trim_trailing_slashes_and_curdir(&p[..idx]);
+                    if let Some(res) = unix_bytes_to_path(parent_bytes) {
+                        return Some(res);
+                    }
+                }
+            }
+        }
+    } else if p != b"." && p != b".." {
+        return Some(Path::new(""));
+    }
+    path.parent()
+}
+
+/// Splits a Unix `Path` into `(parent, file_name)` in a single reverse scan.
+#[inline]
+pub fn unix_split_parent_and_file_name(path: &Path) -> Option<(&Path, &std::ffi::OsStr)> {
+    let p = path.as_os_str().as_encoded_bytes();
+    if p.is_empty() || p == b"/" {
+        return None;
+    }
+    if let Some(idx) = p.iter().rposition(|&b| b == b'/') {
+        if idx + 1 < p.len() {
+            let tail = &p[idx + 1..];
+            if tail != b"." && tail != b".." {
+                if idx == 0 {
+                    if let Some(tail_os) = unix_bytes_to_os_str(tail) {
+                        return Some((Path::new("/"), tail_os));
+                    }
+                } else if !(idx == 1 && p[0] == b'.') {
+                    let parent_bytes = trim_trailing_slashes_and_curdir(&p[..idx]);
+                    if let (Some(parent_p), Some(tail_os)) = (unix_bytes_to_path(parent_bytes), unix_bytes_to_os_str(tail)) {
+                        return Some((parent_p, tail_os));
+                    }
+                }
+            }
+        }
+    } else if p != b"."
+        && p != b".."
+        && let Some(tail_os) = unix_bytes_to_os_str(p)
+    {
+        return Some((Path::new(""), tail_os));
+    }
+    Some((path.parent()?, path.file_name()?))
+}
+
+/// Verifies that `path.parent() == Some(expected_parent)` and returns `path.file_name()` in one
+/// slice prefix check.
+#[inline]
+pub fn unix_direct_child_file_name<'a>(path: &'a Path, expected_parent: &Path) -> Option<&'a std::ffi::OsStr> {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(expected_parent.as_os_str().as_encoded_bytes());
+    if !b.is_empty() && b != b"." && !b.starts_with(b"./") && !unix_path_has_dotdot_segment(b) {
+        if b == b"/" {
+            if p.starts_with(b"/") {
+                let rem = strip_leading_slashes_and_curdir(&p[1..]);
+                if !rem.is_empty()
+                    && rem != b"."
+                    && rem != b".."
+                    && !rem.contains(&b'/')
+                    && let Some(res) = unix_bytes_to_os_str(rem)
+                {
+                    return Some(res);
+                }
+            }
+        } else if p.starts_with(b) && p.len() > b.len() + 1 && p[b.len()] == b'/' {
+            let rem = strip_leading_slashes_and_curdir(&p[b.len() + 1..]);
+            if !rem.is_empty()
+                && rem != b"."
+                && rem != b".."
+                && !rem.contains(&b'/')
+                && let Some(res) = unix_bytes_to_os_str(rem)
+            {
+                return Some(res);
+            }
+        }
+    }
+    if path.parent() != Some(expected_parent) {
+        return None;
+    }
+    path.file_name()
+}
+
 /// Returns `true` if `path` contains a `..` component (parent directory traversal).
 fn contains_parent_dir_component(path: &str) -> bool {
    path.split(['/', '\\']).any(|component| component == "..")
@@ -1211,4 +1563,110 @@ mod tests {
        assert!(normalize_extract_entry_key("file.txt", Some("../bad"), false).is_err());
        assert!(normalize_extract_entry_key("/absolute", None, false).is_err());
    }
+
+    #[test]
+    fn unix_swar_path_helpers_match_std_path_across_boundaries() {
+        for pad in 1..=20 {
+            let prefix = "a".repeat(pad);
+            let clean_cases = [
+                format!("/{prefix}/bucket/object/xl.meta"),
+                format!("/{prefix}/.rustfs.sys/tmp/uuid/part.1"),
+                format!("/{prefix}/dir..name/..hidden/file...ext"),
+                format!("{prefix}/relative/object.tar.gz"),
+            ];
+            for case in &clean_cases {
+                let bytes = case.as_bytes();
+                assert!(!unix_path_has_dotdot_segment(bytes), "clean case flagged dotdot: {case}");
+                assert!(unix_is_canonical_ascii_path(bytes), "canonical case rejected: {case}");
+            }
+
+            let non_canonical_slash_cases = [
+                format!("//{prefix}/bucket/object/xl.meta"),
+                format!("/{prefix}//bucket/object/xl.meta"),
+                format!("/{prefix}/bucket/object/"),
+            ];
+            for case in &non_canonical_slash_cases {
+                assert!(
+                    !unix_is_canonical_ascii_path(case.as_bytes()),
+                    "non-canonical slash case accepted: {case}"
+                );
+            }
+
+            let dot_cases = [
+                format!("/{prefix}/./xl.meta"),
+                format!("/{prefix}/bucket/."),
+                format!("./{prefix}/bucket"),
+                ".".to_string(),
+            ];
+            for case in &dot_cases {
+                let bytes = case.as_bytes();
+                assert!(!unix_path_has_dotdot_segment(bytes), "dot-only case flagged as dotdot: {case}");
+                assert!(!unix_is_canonical_ascii_path(bytes), "dot case accepted as canonical: {case}");
+            }
+
+            let dotdot_cases = [
+                format!("/{prefix}/../xl.meta"),
+                format!("/{prefix}/bucket/.."),
+                format!("../{prefix}/bucket"),
+                "..".to_string(),
+            ];
+            for case in &dotdot_cases {
+                let bytes = case.as_bytes();
+                assert!(unix_path_has_dotdot_segment(bytes), "dotdot case missed by dotdot: {case}");
+                assert!(!unix_is_canonical_ascii_path(bytes), "dotdot case accepted as canonical: {case}");
+            }
+        }
+
+        let base = Path::new("/tmp/rustfs-disk1");
+        let child = Path::new("/tmp/rustfs-disk1/bucket/prefix/xl.meta");
+        let sibling = Path::new("/tmp/rustfs-disk10/bucket/prefix/xl.meta");
+        assert!(unix_normalized_path_starts_with(child, base));
+        assert!(!unix_normalized_path_starts_with(sibling, base));
+        assert_eq!(unix_normalized_path_strip_prefix(child, base), Some(Path::new("bucket/prefix/xl.meta")));
+        assert_eq!(unix_normalized_path_strip_prefix(base, base), Some(Path::new("")));
+        assert_eq!(unix_normalized_path_strip_prefix(sibling, base), None);
+        assert_eq!(unix_path_parent(child), Some(Path::new("/tmp/rustfs-disk1/bucket/prefix")));
+        assert_eq!(unix_path_parent(Path::new("/")), None);
+        assert_eq!(unix_path_parent(Path::new("/rootchild")), Some(Path::new("/")));
+        assert_eq!(unix_path_parent(Path::new("relative")), Some(Path::new("")));
+        assert_eq!(
+            unix_split_parent_and_file_name(child),
+            Some((Path::new("/tmp/rustfs-disk1/bucket/prefix"), std::ffi::OsStr::new("xl.meta")))
+        );
+        assert_eq!(
+            unix_direct_child_file_name(child, Path::new("/tmp/rustfs-disk1/bucket/prefix")),
+            Some(std::ffi::OsStr::new("xl.meta"))
+        );
+        assert_eq!(unix_direct_child_file_name(child, base), None);
+
+        let proc_root = Path::new("/proc/self/fd/12/.");
+        let proc_bucket = Path::new("/proc/self/fd/12/./bucket");
+        let proc_obj = Path::new("/proc/self/fd/12/./bucket/prefix/xl.meta");
+        assert!(unix_normalized_path_starts_with(proc_bucket, proc_root));
+        assert!(unix_normalized_path_starts_with(proc_obj, proc_bucket));
+        assert!(unix_normalized_path_starts_with(proc_obj, proc_root));
+        assert_eq!(
+            unix_normalized_path_strip_prefix(proc_obj, proc_root),
+            Some(Path::new("bucket/prefix/xl.meta"))
+        );
+        assert_eq!(
+            unix_normalized_path_strip_prefix(proc_obj, proc_bucket),
+            Some(Path::new("prefix/xl.meta"))
+        );
+        assert_eq!(unix_path_parent(proc_obj), Some(Path::new("/proc/self/fd/12/./bucket/prefix")));
+        assert_eq!(
+            unix_direct_child_file_name(proc_obj, Path::new("/proc/self/fd/12/./bucket/prefix")),
+            Some(std::ffi::OsStr::new("xl.meta"))
+        );
+
+        assert!(unix_is_canonical_ascii_path("bucket/对象-ключ-é/xl.meta".as_bytes()));
+        assert!(unix_normalized_path_starts_with(
+            Path::new("/data/rustfs/bucket/dir/xl.meta"),
+            Path::new("/data/rustfs/bucket//dir")
+        ));
+        assert_eq!(
+            unix_normalized_path_strip_prefix(Path::new("/a/b/"), Path::new("/a")),
+            Some(Path::new("b"))
+        );
+    }
 }