feat(rustfs): opt-in MD5 lane server via RUSTFS_MD5_LANE_SERVER

Feature hash-md5-simd builds the multi-stream engine into the binary. At
startup, RUSTFS_MD5_LANE_SERVER=on installs a LaneServer as the ETag hashing
sink (RUSTFS_MD5_LANE_THREADS, RUSTFS_MD5_LANE_MIN_STREAMS tune it); anything
else keeps inline hashing, and the default build does not compile the path.

Smoke-tested on a local 4-volume instance with the server on: 32 single-part
objects across padding boundaries (0..7 MB) and 60 concurrent 2 MiB uploads
all have ETag == MD5; a 40 MiB multipart object matches the S3 part-digest
rule and downloads intact.

.github/workflows/ci.yml

@@ -514,9 +514,18 @@ jobs:
       matrix:
         features:
           - name: swift
+            packages: "-p rustfs -p rustfs-protocols"
             flags: "--features swift"
           - name: sftp
+            packages: "-p rustfs -p rustfs-protocols"
             flags: "--features sftp"
+          # Alternate MD5 backend behind rustfs_utils::hash::Md5Stream. Not a
+          # protocol, but the same shape: a build feature that must keep the
+          # crates it touches green. utils carries the backend and its
+          # differential oracle, rio carries the ETag / Content-MD5 consumers.
+          - name: hash-md5-fast-md5
+            packages: "-p rustfs-utils -p rustfs-rio"
+            flags: "--features rustfs-utils/hash-md5-fast-md5,rustfs-rio/hash-md5-fast-md5"
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
@@ -538,7 +547,7 @@ jobs:
 
       - name: Run clippy with ${{ matrix.features.name }}
         run: |
-          cargo clippy -p rustfs -p rustfs-protocols --all-targets ${{ matrix.features.flags }} -- -D warnings
+          cargo clippy ${{ matrix.features.packages }} --all-targets ${{ matrix.features.flags }} -- -D warnings
 
       - name: Run tests with ${{ matrix.features.name }}
         env:
@@ -549,7 +558,7 @@ jobs:
           # --profile ci so the quarantine list (and its junit flaky markers)
           # covers this leg too; the default profile is the local no-retry
           # profile and silently ignored quarantined flakes here (rustfs#6703).
-          cargo nextest run --profile ci -p rustfs -p rustfs-protocols ${{ matrix.features.flags }}
+          cargo nextest run --profile ci ${{ matrix.features.packages }} ${{ matrix.features.flags }}
 
   build-rustfs-debug-binary:
     name: Build RustFS Debug Binary

Cargo.lock

@@ -4307,6 +4307,12 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
 
+[[package]]
+name = "fast-md5"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3d5eae88ed9322ba43db8f1ce3214050d9dea7730f44b2de0769604526b8d584"
+
 [[package]]
 name = "faster-hex"
 version = "0.10.0"
@@ -6409,6 +6415,11 @@ version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c"
 
+[[package]]
+name = "md5-simd"
+version = "0.2.0"
+source = "git+https://github.com/reatang/md5-simd?rev=a202406#a2024068d11f5879036138aedb9e6612f05688c5"
+
 [[package]]
 name = "memchr"
 version = "2.8.3"
@@ -10447,7 +10458,6 @@ dependencies = [
  "hyper-util",
  "ipnetwork",
  "libunftp",
- "md-5 0.11.0",
  "percent-encoding",
  "proptest",
  "regex",
@@ -10560,6 +10570,7 @@ dependencies = [
  "rustfs-io-metrics",
  "rustfs-tls-runtime",
  "rustfs-utils",
+ "rustix",
  "rustls-pki-types",
  "serde",
  "serde_json",
@@ -10614,7 +10625,6 @@ dependencies = [
  "hyper-rustls",
  "hyper-util",
  "lazy_static",
- "md-5 0.11.0",
  "quick-xml",
  "rand 0.10.2",
  "rustfs-checksums",
@@ -10974,6 +10984,7 @@ dependencies = [
  "convert_case 0.12.0",
  "crc-fast",
  "criterion",
+ "fast-md5",
  "flate2",
  "futures",
  "hex-simd",
@@ -10985,6 +10996,7 @@ dependencies = [
  "local-ip-address",
  "lz4",
  "md-5 0.11.0",
+ "md5-simd",
  "netif",
  "proptest",
  "regex",

Cargo.toml

@@ -287,6 +287,12 @@ memmap2 = "0.9.11"
 lz4 = "1.28.1"
 matchit = "0.9.2"
 md-5 = "0.11.0"
+# MD5 backend selectable with the `hash-md5-fast-md5` feature (crates/utils/src/hash_stream.rs).
+# Pinned exactly: single-maintainer crate with hand-written assembly; bump deliberately.
+fast-md5 = "=1.0.0"
+# Multi-stream SIMD MD5 (incremental `update_many`), from the reatang fork until
+# houseme/md5-simd#1 is merged and released; then switch to the crates.io version.
+md5-simd = { git = "https://github.com/reatang/md5-simd", rev = "a202406", default-features = false, features = ["std", "opt", "simd"] }
 mime_guess = "2.0.5"
 moka = { version = "0.12.16" }
 netif = "0.1.6"

crates/ecstore/Cargo.toml

@@ -196,7 +196,6 @@ async-channel.workspace = true
 num_cpus = { workspace = true }
 rand = { workspace = true, features = ["serde"] }
 pin-project-lite.workspace = true
-md-5.workspace = true
 memmap2 = { workspace = true }
 libc.workspace = true
 # "process" adds getrlimit for the io_uring fd-cache RLIMIT_NOFILE gate
@@ -246,6 +245,8 @@ windows-sys = { workspace = true, features = [
 windows-sys = { workspace = true, features = ["Win32_System_Ioctl"] }
 
 [dev-dependencies]
+# Independent MD5 reference for tests; production code goes through rustfs-utils.
+md-5.workspace = true
 aws-smithy-async.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util", "fs"] }
 criterion = { workspace = true, features = ["html_reports"] }

crates/ecstore/src/set_disk/mod.rs

@@ -119,7 +119,6 @@ use bytes::Bytes;
 use bytesize::ByteSize;
 use futures::future::join_all;
 use http::HeaderMap;
-use md5::{Digest as Md5Digest, Md5};
 use regex::Regex;
 use rustfs_config::MI_B;
 use rustfs_filemeta::metadata_keys;
@@ -6980,11 +6979,8 @@ fn get_complete_multipart_md5(parts: &[CompletePart]) -> String {
         }
     }
 
-    let mut hasher = Md5::new();
-    hasher.update(&buf);
-
-    let digest = hasher.finalize();
-    let etag_hex = faster_hex::hex_string(digest.as_slice());
+    let digest = rustfs_utils::hash_stream::Md5Stream::digest(&buf);
+    let etag_hex = faster_hex::hex_string(&digest);
     format!("{}-{}", etag_hex, parts.len())
 }
 

crates/kms/Cargo.toml

@@ -64,9 +64,8 @@ tempfile = { workspace = true }
 moka = { workspace = true, features = ["future"] }
 
 # Additional dependencies
-md-5 = { workspace = true }
 arc-swap = { workspace = true }
-rustfs-utils = { workspace = true, features = ["http"] }
+rustfs-utils = { workspace = true, features = ["http", "hash"] }
 rustfs-security-governance = { workspace = true }
 # `EventName` for KMS audit records. A leaf crate with no rustfs dependencies,
 # so the audit sink can live outside this crate without a second, drifting
@@ -95,6 +94,8 @@ aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] }
 aws-smithy-types = { workspace = true }
 
 [dev-dependencies]
+# Independent MD5 reference for behaviour tests; production code goes through rustfs-utils.
+md-5 = { workspace = true }
 anyhow = { workspace = true }
 # Debugging recorder for asserting emitted metrics in tests.
 metrics-util = { workspace = true, features = ["debugging"] }

crates/kms/src/service.rs

@@ -24,8 +24,8 @@ use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
 use crate::types::*;
 use jiff::Zoned;
-use md5::{Digest as Md5Digest, Md5};
 use rand::random;
+use rustfs_utils::hash_stream::Md5Stream;
 use rustfs_utils::http::object_encryption_keys::{
     INTERNAL_ENCRYPTION_ALGORITHM_HEADER, INTERNAL_ENCRYPTION_CONTEXT_HEADER, INTERNAL_ENCRYPTION_IV_HEADER,
     INTERNAL_ENCRYPTION_KEY_HEADER, INTERNAL_ENCRYPTION_KEY_ID_HEADER, INTERNAL_ENCRYPTION_TAG_HEADER,
@@ -37,9 +37,7 @@ use tracing::debug;
 use zeroize::Zeroize;
 
 fn md5_hex(input: impl AsRef<[u8]>) -> String {
-    let mut hasher = Md5::new();
-    hasher.update(input.as_ref());
-    hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower)
+    hex_simd::encode_to_string(Md5Stream::digest(input.as_ref()), hex_simd::AsciiCase::Lower)
 }
 
 /// Data key for object encryption

crates/protocols/Cargo.toml

@@ -96,7 +96,7 @@ swift = [
     "dep:tokio-util",
     "dep:serde",
     "dep:urlencoding",
-    "dep:md-5",
+    "rustfs-utils/hash",
     "dep:hmac",
     "dep:sha1",
     "dep:hex-simd",
@@ -160,7 +160,6 @@ http-body-util = { workspace = true, optional = true }
 tokio-util = { workspace = true, optional = true, features = ["rt", "io", "compat"] }
 serde = { workspace = true, optional = true, features = ["derive"] }
 urlencoding = { workspace = true, optional = true }
-md-5 = { workspace = true, optional = true }
 hmac = { workspace = true, optional = true }
 sha1 = { workspace = true, optional = true }
 hex-simd = { workspace = true, optional = true }

crates/protocols/src/swift/slo.rs

@@ -21,8 +21,8 @@
 use super::storage_api::large_object::HTTPRangeSpec;
 use super::{SwiftError, object};
 use axum::http::{HeaderMap, Response, StatusCode};
-use md5::{Digest as Md5Digest, Md5};
 use rustfs_credentials::Credentials;
+use rustfs_utils::hash_stream::Md5Stream;
 use s3s::Body;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -82,11 +82,9 @@ impl SLOManifest {
             etag_concat.push_str(etag);
         }
 
-        let mut hasher = Md5::new();
-        hasher.update(etag_concat.as_bytes());
         format!(
             "\"{}-{}\"",
-            hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower),
+            hex_simd::encode_to_string(Md5Stream::digest(etag_concat.as_bytes()), hex_simd::AsciiCase::Lower),
             self.segments.len()
         )
     }

crates/rio/Cargo.toml

@@ -30,6 +30,10 @@ workspace = true
 
 [features]
 default = []
+# Forwarded so `-p rustfs-rio --features hash-md5-fast-md5` tests rio on the alternate MD5 backend.
+hash-md5-fast-md5 = ["rustfs-utils/hash-md5-fast-md5"]
+# Multi-stream MD5 kernel for `md5_lanes::LaneServer` (trial wiring, see rustfs-utils).
+hash-md5-simd = ["rustfs-utils/hash-md5-simd"]
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",
@@ -78,7 +82,6 @@ rustfs-io-metrics.workspace = true
 rustfs-tls-runtime.workspace = true
 rustfs-utils = { workspace = true, features = ["io", "hash", "compress"] }
 serde_json = { workspace = true, features = ["raw_value"] }
-md-5 = { workspace = true }
 minlz.workspace = true
 tracing.workspace = true
 thiserror.workspace = true
@@ -89,6 +92,10 @@ xxhash-rust = { workspace = true, features = ["xxh64", "xxh3"] }
 hex-simd.workspace = true
 
 [dev-dependencies]
+# Independent MD5 reference for etag/checksum tests; production code goes through rustfs-utils.
+md-5 = { workspace = true }
+# Process CPU time for the ignored `etag_throughput` benchmark.
+rustix = { workspace = true, features = ["time"] }
 temp-env = { workspace = true, features = ["async_closure"] }
 tokio = { workspace = true, features = ["test-util"] }
 tokio-test = { workspace = true }

crates/rio/src/checksum.rs

@@ -1080,8 +1080,11 @@ 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.
+/// MD5 hasher for the `x-amz-checksum-*` path. Backed by the shared
+/// `rustfs_utils::hash::Md5Stream`, so the ETag path and this path can never
+/// disagree about which MD5 implementation is in use.
 pub struct Md5Hasher {
-    hasher: md5::Md5,
+    hasher: rustfs_utils::hash::Md5Stream,
 }
 
 impl Default for Md5Hasher {
@@ -1092,14 +1095,14 @@ impl Default for Md5Hasher {
 
 impl Md5Hasher {
     pub fn new() -> Self {
-        use md5::Digest as _;
-        Self { hasher: md5::Md5::new() }
+        Self {
+            hasher: rustfs_utils::hash::Md5Stream::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())
     }
@@ -1110,14 +1113,17 @@ impl Write for Md5Hasher {
 }
 
 impl ChecksumHasher for Md5Hasher {
+    /// Returns the digest of everything written so far and restarts the
+    /// hasher, i.e. `finalize` followed by `reset`. `HashReader` calls this
+    /// exactly once at EOF, so the implicit reset is unobservable there; a
+    /// caller that keeps writing afterwards gets a digest of the new bytes
+    /// only, which is the same contract `reset` documents.
     fn finalize(&mut self) -> Vec<u8> {
-        use md5::Digest as _;
-        self.hasher.clone().finalize().to_vec()
+        self.hasher.finalize_reset().to_vec()
     }
 
     fn reset(&mut self) {
-        use md5::Digest as _;
-        self.hasher = md5::Md5::new();
+        self.hasher = rustfs_utils::hash::Md5Stream::new();
     }
 }
 
@@ -1660,6 +1666,38 @@ mod tests {
         }
     }
 
+    /// `Md5Hasher` is the `x-amz-checksum-md5` shell over `Md5Stream`; pin the
+    /// `finalize`-restarts contract that `HashReader` relies on at EOF.
+    #[test]
+    fn md5_hasher_finalize_restarts_and_reset_clears() {
+        use super::ChecksumHasher as _;
+        use std::io::Write as _;
+        fn reference(data: &[u8]) -> Vec<u8> {
+            <md5::Md5 as md5::Digest>::digest(data).to_vec()
+        }
+        let body: Vec<u8> = (0..70_000u32).map(|i| (i * 7 % 251) as u8).collect();
+        let want = reference(&body);
+
+        let mut hasher = super::Md5Hasher::default();
+        for chunk in body.chunks(1234) {
+            hasher.write_all(chunk).expect("in-memory write");
+        }
+        hasher.flush().expect("flush is a no-op");
+        assert_eq!(hasher.finalize(), want);
+        // finalize restarted the hasher: a second call is the digest of nothing.
+        assert_eq!(hasher.finalize(), reference(b""));
+
+        hasher.write_all(b"partial").expect("in-memory write");
+        hasher.reset();
+        hasher.write_all(&body).expect("in-memory write");
+        assert_eq!(hasher.finalize(), want);
+
+        // The trait-object path used by HashReader resolves to the same shell.
+        let mut boxed = ChecksumType::MD5.hasher().expect("MD5 has a hasher");
+        boxed.write_all(&body).expect("in-memory write");
+        assert_eq!(boxed.finalize(), want);
+    }
+
     // Drift lock for the backlog#1844 PR3 verdict: rio keeps its own hasher
     // shells instead of delegating to rustfs-checksums, so both sides pin the
     // SAME input and official digests (crates/checksums/src/lib.rs pins these
@@ -1673,6 +1711,9 @@ mod tests {
             (ChecksumType::CRC64_NVME, "aecaf3af9c98a855"),
             (ChecksumType::SHA1, "f48dd853820860816c75d54d0f584dc863327a7c"),
             (ChecksumType::SHA256, "916f0027a575074ce72a331777c3478d6513f786a591bd892da1a577bf2335f9"),
+            // Content-MD5 shares this vector with crates/checksums (`test_md5_checksum`)
+            // and crates/utils (`test_hash_encode_md5`).
+            (ChecksumType::MD5, "eb733a00c0c9d336e65691a37ab54293"),
         ] {
             assert_eq!(raw_hex(t, b"test data"), want_hex, "{t:?} digest drifted from the shared vector");
         }

crates/rio/src/etag_reader.rs

@@ -13,11 +13,11 @@
 // limitations under the License.
 
 use crate::compress_index::{Index, TryGetIndex};
+use crate::md5_lanes::{Md5Lane, Md5StreamSink, md5_stream_sink};
 use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
-use md5::{Digest, Md5};
 use pin_project_lite::pin_project;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{Context, Poll, ready};
 use tokio::io::{AsyncRead, ReadBuf};
 use tracing::error;
 
@@ -25,7 +25,11 @@ pin_project! {
     pub struct  EtagReader<R> {
         #[pin]
         pub inner: R,
-        pub md5: Md5,
+        // Where the plaintext goes to be hashed. `Some` until the digest has been taken.
+        // The digest exists only after EOF: there is no snapshot of a stream in progress.
+        lane: Option<Box<dyn Md5Lane>>,
+        // The inner reader reported EOF; from here on only the digest is awaited.
+        inner_eof: bool,
         pub finished: bool,
         pub checksum: Option<String>,
         resolved_etag: Option<String>,
@@ -33,27 +37,30 @@ pin_project! {
 }
 
 impl<R> EtagReader<R> {
+    /// Hash through the process-wide [`Md5StreamSink`] (inline unless one was installed).
     pub fn new(inner: R, checksum: Option<String>) -> Self {
+        Self::with_sink(inner, checksum, md5_stream_sink().as_ref())
+    }
+
+    /// Hash through `sink`. Tests and benchmarks use this to stay off the process-wide sink.
+    pub fn with_sink(inner: R, checksum: Option<String>, sink: &dyn Md5StreamSink) -> Self {
         Self {
             inner,
-            md5: Md5::new(),
+            lane: Some(sink.open()),
+            inner_eof: false,
             finished: false,
             checksum,
             resolved_etag: None,
         }
     }
 
-    /// 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 {
-        if let Some(etag) = &self.resolved_etag {
-            return etag.clone();
-        }
-
-        let etag = self.md5.clone().finalize().to_vec();
-        let etag = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower);
-        self.resolved_etag = Some(etag.clone());
-        etag
+    /// The final md5 value (etag) as a hex string.
+    ///
+    /// `None` until the inner stream has reached EOF: the hasher is consumed
+    /// exactly once at that point, so there is no partial digest to hand out
+    /// earlier. After EOF this returns the same cached value every time.
+    pub fn get_etag(&self) -> Option<String> {
+        self.resolved_etag.clone()
     }
 }
 
@@ -66,40 +73,46 @@ where
         if *this.finished {
             return Poll::Ready(Ok(()));
         }
+        // `lane` is `Some` until `finished`; treat the impossible `None` as EOF rather than
+        // panicking inside the data path.
+        let Some(lane) = this.lane.as_mut() else {
+            *this.finished = true;
+            return Poll::Ready(Ok(()));
+        };
 
-        let orig_filled = buf.filled().len();
-        let poll = this.inner.as_mut().poll_read(cx, buf);
-        if let Poll::Ready(Ok(())) = &poll {
+        if !*this.inner_eof {
+            // Do not pull more plaintext while the hasher is behind on what it already has.
+            ready!(lane.poll_ready(cx));
+            let orig_filled = buf.filled().len();
+            ready!(this.inner.as_mut().poll_read(cx, buf))?;
             let filled = &buf.filled()[orig_filled..];
             if !filled.is_empty() {
-                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
-                    && *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,
-                        },
-                    )));
-                }
+                lane.write(filled);
+                return Poll::Ready(Ok(()));
             }
+            *this.inner_eof = true;
         }
-        poll
+
+        // EOF: nothing was added to `buf` on this path, so resolving late still reads as EOF.
+        let digest = ready!(lane.poll_finish(cx));
+        *this.finished = true;
+        *this.lane = None;
+        let etag = hex_simd::encode_to_string(digest?, hex_simd::AsciiCase::Lower);
+        *this.resolved_etag = Some(etag.clone());
+
+        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,
+                },
+            )));
+        }
+        Poll::Ready(Ok(()))
     }
 }
 
@@ -112,7 +125,7 @@ impl<R> EtagResolvable for EtagReader<R> {
         if let Some(checksum) = &self.checksum {
             Some(checksum.clone())
         } else if self.finished {
-            Some(self.get_etag())
+            self.get_etag()
         } else {
             None
         }
@@ -144,6 +157,9 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    // RustCrypto md-5 stays a dev-dependency: the expected values must come
+    // from an implementation independent of the one under test.
+    use md5::{Digest, Md5};
     use rand::RngExt;
     use std::io::Cursor;
     use tokio::io::{AsyncReadExt, BufReader};
@@ -216,6 +232,54 @@ mod tests {
         let mut buf = [0u8; 2];
         let _ = etag_reader.read(&mut buf).await.unwrap();
         assert_eq!(etag_reader.try_resolve_etag(), None);
+        assert_eq!(etag_reader.get_etag(), None, "no partial digest before EOF");
+
+        // Reading the rest resolves it, and the value is stable afterwards.
+        let mut rest = Vec::new();
+        etag_reader.read_to_end(&mut rest).await.unwrap();
+        let expected = faster_hex::hex_string(Md5::digest(data).as_slice());
+        assert_eq!(etag_reader.get_etag(), Some(expected.clone()));
+        assert_eq!(etag_reader.try_resolve_etag(), Some(expected));
+    }
+
+    /// The body stream never hands EtagReader the object in one piece; the
+    /// digest must not depend on how the inner reader splits its reads.
+    #[tokio::test]
+    async fn test_etag_reader_small_inner_reads_match_one_shot() {
+        let size = 3 * 64 * 1024 + 77;
+        let mut data = vec![0u8; size];
+        rand::rng().fill(&mut data[..]);
+        let expected = faster_hex::hex_string(Md5::digest(&data).as_slice());
+
+        // BufReader with a tiny capacity forces many short poll_read fills.
+        let inner = BufReader::with_capacity(61, Cursor::new(data.clone()));
+        let mut etag_reader = EtagReader::new(inner, None);
+        let mut out = Vec::new();
+        let mut chunk = [0u8; 61];
+        loop {
+            let n = etag_reader.read(&mut chunk).await.unwrap();
+            if n == 0 {
+                break;
+            }
+            out.extend_from_slice(&chunk[..n]);
+        }
+        assert_eq!(out, data);
+        assert_eq!(etag_reader.try_resolve_etag(), Some(expected));
+    }
+
+    /// Reads after EOF are a no-op and never disturb the resolved etag.
+    #[tokio::test]
+    async fn test_etag_reader_reads_after_eof_are_stable() {
+        let data = b"stable after eof";
+        let expected = faster_hex::hex_string(Md5::digest(data).as_slice());
+        let mut etag_reader = EtagReader::new(BufReader::new(&data[..]), None);
+        let mut buf = Vec::new();
+        etag_reader.read_to_end(&mut buf).await.unwrap();
+        for _ in 0..3 {
+            let mut extra = [0u8; 8];
+            assert_eq!(etag_reader.read(&mut extra).await.unwrap(), 0);
+            assert_eq!(etag_reader.get_etag(), Some(expected.clone()));
+        }
     }
 
     #[tokio::test]
@@ -274,6 +338,87 @@ mod tests {
             .and_then(|source| source.downcast_ref::<BadDigest>())
             .expect("checksum mismatch should preserve the BadDigest type");
         assert_eq!(digest.expected_md5, wrong_checksum);
-        assert_eq!(digest.calculated_md5, calculated_md5);
+        assert_eq!(digest.calculated_md5, calculated_md5.clone());
+        // The digest was resolved before the comparison, so it stays observable
+        // after the error and the reader stays at EOF.
+        assert_eq!(etag_reader.get_etag(), Some(calculated_md5));
+        assert!(etag_reader.finished);
+    }
+
+    /// An inner reader that yields part of the body, then one `Err`, then the
+    /// rest. EtagReader must surface the error, must not treat it as EOF, and
+    /// must keep hashing correctly once the inner reader recovers.
+    struct FlakyInner {
+        data: Vec<u8>,
+        pos: usize,
+        fail_at: usize,
+        failed: bool,
+    }
+
+    impl AsyncRead for FlakyInner {
+        fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if self.pos >= self.fail_at && !self.failed {
+                self.failed = true;
+                return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "transient")));
+            }
+            let end = (self.pos + buf.remaining().min(7)).min(self.data.len());
+            buf.put_slice(&self.data[self.pos..end]);
+            self.pos = end;
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn test_etag_reader_inner_error_is_not_eof_and_state_survives() {
+        let data: Vec<u8> = (0..1000u32).map(|i| (i * 31 % 251) as u8).collect();
+        let expected = faster_hex::hex_string(Md5::digest(&data).as_slice());
+        let mut etag_reader = EtagReader::new(
+            FlakyInner {
+                data: data.clone(),
+                pos: 0,
+                fail_at: 300,
+                failed: false,
+            },
+            None,
+        );
+
+        let mut out = Vec::new();
+        let mut chunk = [0u8; 64];
+        let mut saw_error = false;
+        loop {
+            match etag_reader.read(&mut chunk).await {
+                Ok(0) => break,
+                Ok(n) => out.extend_from_slice(&chunk[..n]),
+                Err(err) => {
+                    assert_eq!(err.kind(), std::io::ErrorKind::Interrupted);
+                    assert!(!etag_reader.finished, "an inner error must not finish the reader");
+                    assert_eq!(etag_reader.get_etag(), None);
+                    saw_error = true;
+                }
+            }
+        }
+        assert!(saw_error, "the inner reader must have failed once");
+        assert_eq!(out, data);
+        assert_eq!(etag_reader.try_resolve_etag(), Some(expected));
+    }
+
+    /// Interleaved `Pending` from the inner reader (a slow network body) must
+    /// neither be treated as data nor as EOF.
+    #[tokio::test]
+    async fn test_etag_reader_survives_pending_between_chunks() {
+        let data = b"pending between chunks keeps the digest intact";
+        let expected = faster_hex::hex_string(Md5::digest(data).as_slice());
+        let inner = tokio_test::io::Builder::new()
+            .read(&data[..10])
+            .wait(std::time::Duration::from_millis(5))
+            .read(&data[10..20])
+            .wait(std::time::Duration::from_millis(5))
+            .read(&data[20..])
+            .build();
+        let mut etag_reader = EtagReader::new(inner, None);
+        let mut out = Vec::new();
+        etag_reader.read_to_end(&mut out).await.unwrap();
+        assert_eq!(out, data);
+        assert_eq!(etag_reader.try_resolve_etag(), Some(expected));
     }
 }

crates/rio/src/hash_reader.rs

@@ -778,6 +778,52 @@ mod tests {
         assert_eq!(buf, data);
     }
 
+    /// Content-MD5 through the full HashReader stack: a matching value is
+    /// accepted and the resolved etag is the independent MD5 of the body.
+    #[tokio::test]
+    async fn content_md5_match_through_hash_reader_resolves_etag() {
+        use md5::Digest as _;
+        let data: Vec<u8> = (0..(256 * 1024 + 9)).map(|i| (i * 13 % 251) as u8).collect();
+        let expected_md5 = faster_hex::hex_string(md5::Md5::digest(&data).as_slice());
+        let reader = BufReader::with_capacity(4096, Cursor::new(data.clone()));
+        let mut hash_reader =
+            HashReader::from_stream(reader, data.len() as i64, data.len() as i64, Some(expected_md5.clone()), None, false)
+                .expect("operation should succeed");
+        let mut buf = Vec::new();
+        hash_reader
+            .read_to_end(&mut buf)
+            .await
+            .expect("matching Content-MD5 must be accepted");
+        assert_eq!(buf, data);
+        assert_eq!(hash_reader.try_resolve_etag(), Some(expected_md5));
+    }
+
+    /// Content-MD5 mismatch must surface as the typed `BadDigest` that the API
+    /// layer maps to the S3 `BadDigest` error, not as an opaque io::Error.
+    #[tokio::test]
+    async fn content_md5_mismatch_through_hash_reader_retains_bad_digest() {
+        use md5::Digest as _;
+        let data = b"tampered content-md5 payload";
+        let wrong_md5 = "0".repeat(32);
+        let calculated = faster_hex::hex_string(md5::Md5::digest(data).as_slice());
+        let reader = BufReader::new(Cursor::new(&data[..]));
+        let mut hash_reader =
+            HashReader::from_stream(reader, data.len() as i64, data.len() as i64, Some(wrong_md5.clone()), None, false)
+                .expect("operation should succeed");
+
+        let error = hash_reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("Content-MD5 mismatch should fail");
+        let digest = error
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<crate::BadDigest>())
+            .expect("Content-MD5 mismatch should remain typed");
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(digest.expected_md5, wrong_md5);
+        assert_eq!(digest.calculated_md5, calculated);
+    }
+
     #[tokio::test]
     async fn sha256_mismatch_retains_typed_io_error_source() {
         let data = b"tampered payload";

crates/rio/src/lib.rs

@@ -106,6 +106,8 @@ pub use s2_decoder::{MAX_S2_DECOMPRESSED_BLOCK_SIZE, S2Decoder};
 mod etag_reader;
 pub use etag_reader::EtagReader;
 
+pub mod md5_lanes;
+
 mod compress_index;
 mod compress_reader;
 pub use compress_reader::{CompressReader, DecompressReader};

crates/rio/src/md5_lanes.rs

@@ -0,0 +1,952 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Where an [`EtagReader`](crate::EtagReader) sends the bytes it sees (Port B).
+//!
+//! MD5 cannot be vectorised within one message, only across independent ones. One upload is one
+//! message, so the only way to use SIMD for ETags is to hash *different requests* together. That
+//! needs a component that outlives any single request: the reader hands its plaintext to a
+//! [`Md5StreamSink`] and asks for the digest at EOF, and never learns how the hashing was done.
+//!
+//! | sink | hashing happens | used when |
+//! |---|---|---|
+//! | [`InlineSink`] | in `poll_read`, on the caller's task, one stream at a time | default; nothing installed |
+//! | [`LaneServer`] | on a dedicated thread, all in-flight uploads advanced together through a [`Md5LaneEngine`] | installed at startup |
+//!
+//! The ETag value, its encoding, `Content-MD5` verification and the moment the ETag becomes
+//! available (EOF) are identical for both.
+
+use bytes::{Bytes, BytesMut};
+use rustfs_utils::hash::Md5Stream;
+use rustfs_utils::hash_lanes::Md5LaneEngine;
+use std::collections::HashMap;
+use std::io;
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, TryRecvError};
+use std::sync::{Arc, Mutex, OnceLock};
+use std::task::{Context, Poll, Waker};
+use std::time::Duration;
+
+/// Opens one [`Md5Lane`] per upload.
+pub trait Md5StreamSink: Send + Sync + 'static {
+    /// Start hashing a new, empty message.
+    fn open(&self) -> Box<dyn Md5Lane>;
+}
+
+/// The MD5 of one upload, fed in order. Dropping a lane before it finishes abandons the message.
+pub trait Md5Lane: Send + Sync {
+    /// Back-pressure: `Pending` while too many of this lane's bytes are still waiting to be
+    /// hashed. Polled before the reader pulls more data from its source.
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<()>;
+
+    /// Append bytes. Never refuses: the reader has already handed these bytes to its caller.
+    fn write(&mut self, data: &[u8]);
+
+    /// No more bytes; resolve to the digest of everything written. Poll until `Ready`.
+    fn poll_finish(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<[u8; 16]>>;
+}
+
+/// Hash in place, one stream at a time. Never pends, never fails.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct InlineSink;
+
+struct InlineLane(Option<Md5Stream>);
+
+impl Md5StreamSink for InlineSink {
+    fn open(&self) -> Box<dyn Md5Lane> {
+        Box::new(InlineLane(Some(Md5Stream::new())))
+    }
+}
+
+impl Md5Lane for InlineLane {
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<()> {
+        Poll::Ready(())
+    }
+
+    fn write(&mut self, data: &[u8]) {
+        if let Some(md5) = self.0.as_mut() {
+            md5.update(data);
+        }
+    }
+
+    fn poll_finish(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<[u8; 16]>> {
+        Poll::Ready(Ok(self.0.take().unwrap_or_default().finalize()))
+    }
+}
+
+static SINK: OnceLock<Arc<dyn Md5StreamSink>> = OnceLock::new();
+
+/// Install the process-wide sink. Call once from the composition root, before serving traffic.
+/// Returns `false` (and changes nothing) if a sink was already installed.
+pub fn install_md5_stream_sink(sink: Arc<dyn Md5StreamSink>) -> bool {
+    SINK.set(sink).is_ok()
+}
+
+/// The installed sink, or [`InlineSink`] when none was installed.
+pub fn md5_stream_sink() -> Arc<dyn Md5StreamSink> {
+    SINK.get().cloned().unwrap_or_else(|| Arc::new(InlineSink))
+}
+
+/// `on` installs a [`LaneServer`]; anything else (default) keeps ETag hashing inline.
+pub const ENV_MD5_LANE_SERVER: &str = "RUSTFS_MD5_LANE_SERVER";
+/// Hashing threads, see [`LaneServerConfig::threads`]. Default 1.
+pub const ENV_MD5_LANE_THREADS: &str = "RUSTFS_MD5_LANE_THREADS";
+/// See [`LaneServerConfig::min_streams`]. Default 4.
+pub const ENV_MD5_LANE_MIN_STREAMS: &str = "RUSTFS_MD5_LANE_MIN_STREAMS";
+
+/// What [`install_lane_server_from_env`] did.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LaneServerInstall {
+    /// Not requested; ETags are hashed inline.
+    Disabled,
+    /// Requested, but this build has no multi-stream engine (`hash-md5-simd`), or the threads
+    /// could not be spawned. ETags are hashed inline.
+    Unavailable,
+    /// A sink was already installed; nothing changed.
+    AlreadyInstalled,
+    /// Installed.
+    Installed {
+        /// Engine name, e.g. `simd-avx2-fused`.
+        engine: &'static str,
+        /// Hashing threads.
+        threads: usize,
+        /// Uploads in flight below which new uploads stay inline.
+        min_streams: usize,
+    },
+}
+
+/// Composition-root entry: install a [`LaneServer`] as the process-wide sink if
+/// [`ENV_MD5_LANE_SERVER`] asks for one. Off by default: with few concurrent uploads and idle
+/// cores, inline hashing on the request tasks has the higher wall-clock throughput; the server
+/// pays off in CPU per byte (roughly a third at 8+ uploads) and in throughput at 16+.
+pub fn install_lane_server_from_env() -> LaneServerInstall {
+    if !rustfs_utils::get_env_str(ENV_MD5_LANE_SERVER, "off").eq_ignore_ascii_case("on") {
+        return LaneServerInstall::Disabled;
+    }
+    #[cfg(not(feature = "hash-md5-simd"))]
+    {
+        LaneServerInstall::Unavailable
+    }
+    #[cfg(feature = "hash-md5-simd")]
+    {
+        let defaults = LaneServerConfig::default();
+        let config = LaneServerConfig {
+            threads: rustfs_utils::get_env_usize(ENV_MD5_LANE_THREADS, defaults.threads).clamp(1, 16),
+            min_streams: rustfs_utils::get_env_usize(ENV_MD5_LANE_MIN_STREAMS, defaults.min_streams).max(1),
+            ..defaults
+        };
+        let Ok(server) = LaneServer::start(rustfs_utils::hash_lanes::SimdLaneEngine::new(), config) else {
+            return LaneServerInstall::Unavailable;
+        };
+        let engine = server.engine_name();
+        if install_md5_stream_sink(server) {
+            LaneServerInstall::Installed {
+                engine,
+                threads: config.threads,
+                min_streams: config.min_streams,
+            }
+        } else {
+            LaneServerInstall::AlreadyInstalled
+        }
+    }
+}
+
+/// Tuning of a [`LaneServer`].
+#[derive(Clone, Copy, Debug)]
+pub struct LaneServerConfig {
+    /// A lane hands bytes to the server once it has this many (always a whole number of MD5
+    /// blocks, so lanes stay block-aligned and can share registers), and at EOF.
+    pub submit_bytes: usize,
+    /// A lane stops its reader while more than this many of its bytes await hashing.
+    pub max_in_flight_bytes: usize,
+    /// With fewer uploads than this in flight, a new upload is hashed inline instead: below the
+    /// engine's SIMD threshold the server only adds a copy and a thread hop.
+    pub min_streams: usize,
+    /// When fewer lanes have data than make a SIMD batch, wait this long for more to arrive.
+    pub batch_wait: Duration,
+    /// Hashing threads. One thread is one core's worth of MD5 (several GiB/s once its registers
+    /// are full); uploads wait on it through back-pressure, so size this above the ingest rate
+    /// the node must sustain. Threads are filled one after the other (see [`LaneServer::open`]),
+    /// so extra threads cost nothing until the load needs them.
+    pub threads: usize,
+}
+
+impl Default for LaneServerConfig {
+    fn default() -> Self {
+        Self {
+            submit_bytes: 256 * 1024,
+            max_in_flight_bytes: 1024 * 1024,
+            min_streams: 4,
+            batch_wait: Duration::from_micros(200),
+            threads: 1,
+        }
+    }
+}
+
+/// Counters of a [`LaneServer`], for metrics and tests.
+#[derive(Debug, Default)]
+pub struct LaneServerStats {
+    /// Uploads hashed by the server.
+    pub lanes_opened: AtomicU64,
+    /// Uploads hashed inline because too few were in flight, or the server had stopped.
+    pub inline_fallbacks: AtomicU64,
+    /// Calls into the engine.
+    pub batches: AtomicU64,
+    /// Sum over all batches of the number of lanes in the batch.
+    pub batched_lanes: AtomicU64,
+}
+
+/// Hashes all in-flight uploads together on dedicated threads.
+///
+/// Dropping the last `Arc<LaneServer>` does not stop uploads already in flight from finishing;
+/// a thread exits once its lanes and the server itself are gone.
+pub struct LaneServer {
+    workers: Vec<WorkerHandle>,
+    config: LaneServerConfig,
+    next_id: AtomicU64,
+    active: Arc<AtomicUsize>,
+    alive: Arc<AtomicBool>,
+    stats: Arc<LaneServerStats>,
+    engine_name: &'static str,
+    /// Lanes one thread takes before the next thread is used.
+    pack_target: usize,
+}
+
+struct WorkerHandle {
+    tx: Mutex<Sender<Msg>>,
+    lanes: Arc<AtomicUsize>,
+}
+
+enum Msg {
+    Open(u64, Arc<Shared>),
+    Chunk(u64, Bytes),
+    Finish(u64),
+    Release(u64),
+}
+
+/// What a lane and the server thread share about one upload.
+#[derive(Default)]
+struct Shared {
+    in_flight: AtomicUsize,
+    outcome: Mutex<Option<Result<[u8; 16], ()>>>,
+    waker: Mutex<Option<Waker>>,
+}
+
+impl Shared {
+    fn wake(&self) {
+        if let Some(waker) = self.waker.lock().unwrap_or_else(|e| e.into_inner()).take() {
+            waker.wake();
+        }
+    }
+
+    fn register(&self, cx: &Context<'_>) {
+        *self.waker.lock().unwrap_or_else(|e| e.into_inner()) = Some(cx.waker().clone());
+    }
+
+    fn settle(&self, outcome: Result<[u8; 16], ()>) {
+        *self.outcome.lock().unwrap_or_else(|e| e.into_inner()) = Some(outcome);
+        self.wake();
+    }
+}
+
+impl LaneServer {
+    /// Spawn the hashing threads.
+    pub fn start<E: Md5LaneEngine + Clone>(engine: E, config: LaneServerConfig) -> io::Result<Arc<Self>> {
+        let alive = Arc::new(AtomicBool::new(true));
+        let stats = Arc::new(LaneServerStats::default());
+        let engine_name = engine.name();
+        let pack_target = engine.preferred_batch().max(1) * 2;
+        let mut workers = Vec::new();
+        for index in 0..config.threads.max(1) {
+            let (tx, rx) = mpsc::channel();
+            let worker = Worker {
+                engine: engine.clone(),
+                rx,
+                lanes: HashMap::new(),
+                batch_wait: config.batch_wait,
+                alive: alive.clone(),
+                stats: stats.clone(),
+            };
+            std::thread::Builder::new()
+                .name(format!("rustfs-md5-lanes-{index}"))
+                .spawn(move || worker.run())?;
+            workers.push(WorkerHandle {
+                tx: Mutex::new(tx),
+                lanes: Arc::new(AtomicUsize::new(0)),
+            });
+        }
+        Ok(Arc::new(Self {
+            workers,
+            config,
+            next_id: AtomicU64::new(0),
+            active: Arc::new(AtomicUsize::new(0)),
+            alive,
+            stats,
+            engine_name,
+            pack_target,
+        }))
+    }
+
+    /// Counters since start.
+    pub fn stats(&self) -> &LaneServerStats {
+        &self.stats
+    }
+
+    /// Name of the engine behind this server (diagnostic).
+    pub fn engine_name(&self) -> &'static str {
+        self.engine_name
+    }
+}
+
+impl Md5StreamSink for LaneServer {
+    fn open(&self) -> Box<dyn Md5Lane> {
+        let guard = ActiveGuard::enter(&self.active);
+        if guard.count < self.config.min_streams || !self.alive.load(Ordering::Acquire) {
+            self.stats.inline_fallbacks.fetch_add(1, Ordering::Relaxed);
+            return Box::new(CountedInlineLane {
+                lane: InlineLane(Some(Md5Stream::new())),
+                _guard: guard,
+            });
+        }
+        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+        let shared = Arc::new(Shared::default());
+        // Pack, then spill: splitting a few uploads across threads would leave every thread's
+        // registers half empty. Only when all threads are packed does the load get balanced.
+        let lanes = |w: &&WorkerHandle| w.lanes.load(Ordering::Acquire);
+        let packing = self.workers.iter().find(|w| lanes(w) < self.pack_target);
+        let Some(worker) = packing.or_else(|| self.workers.iter().min_by_key(lanes)) else {
+            return Box::new(CountedInlineLane {
+                lane: InlineLane(Some(Md5Stream::new())),
+                _guard: guard,
+            });
+        };
+        let tx = worker.tx.lock().unwrap_or_else(|e| e.into_inner()).clone();
+        if tx.send(Msg::Open(id, shared.clone())).is_err() {
+            self.stats.inline_fallbacks.fetch_add(1, Ordering::Relaxed);
+            return Box::new(CountedInlineLane {
+                lane: InlineLane(Some(Md5Stream::new())),
+                _guard: guard,
+            });
+        }
+        self.stats.lanes_opened.fetch_add(1, Ordering::Relaxed);
+        Box::new(ServerLane {
+            id,
+            tx,
+            shared,
+            pending: BytesMut::new(),
+            submit_bytes: self.config.submit_bytes.max(64) & !63,
+            max_in_flight: self.config.max_in_flight_bytes,
+            finish_sent: false,
+            done: false,
+            _guard: guard,
+            _worker_lane: ActiveGuard::enter(&worker.lanes),
+        })
+    }
+}
+
+/// Counts an upload as in flight for as long as its lane exists, inline or not.
+struct ActiveGuard {
+    active: Arc<AtomicUsize>,
+    /// In-flight uploads including this one, at the time it was opened.
+    count: usize,
+}
+
+impl ActiveGuard {
+    fn enter(active: &Arc<AtomicUsize>) -> Self {
+        let count = active.fetch_add(1, Ordering::AcqRel) + 1;
+        Self {
+            active: active.clone(),
+            count,
+        }
+    }
+}
+
+impl Drop for ActiveGuard {
+    fn drop(&mut self) {
+        self.active.fetch_sub(1, Ordering::AcqRel);
+    }
+}
+
+struct CountedInlineLane {
+    lane: InlineLane,
+    _guard: ActiveGuard,
+}
+
+impl Md5Lane for CountedInlineLane {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        self.lane.poll_ready(cx)
+    }
+    fn write(&mut self, data: &[u8]) {
+        self.lane.write(data);
+    }
+    fn poll_finish(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<[u8; 16]>> {
+        self.lane.poll_finish(cx)
+    }
+}
+
+/// The request-side half of an upload hashed by the server.
+struct ServerLane {
+    id: u64,
+    tx: Sender<Msg>,
+    shared: Arc<Shared>,
+    pending: BytesMut,
+    submit_bytes: usize,
+    max_in_flight: usize,
+    finish_sent: bool,
+    done: bool,
+    _guard: ActiveGuard,
+    _worker_lane: ActiveGuard,
+}
+
+impl ServerLane {
+    fn submit(&mut self, len: usize) {
+        if len == 0 {
+            return;
+        }
+        let chunk = self.pending.split_to(len).freeze();
+        self.shared.in_flight.fetch_add(len, Ordering::AcqRel);
+        if self.tx.send(Msg::Chunk(self.id, chunk)).is_err() {
+            self.shared.settle(Err(()));
+        }
+    }
+}
+
+impl Md5Lane for ServerLane {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        if self.shared.in_flight.load(Ordering::Acquire) <= self.max_in_flight {
+            return Poll::Ready(());
+        }
+        self.shared.register(cx);
+        // The server may have caught up between the check and the registration.
+        if self.shared.in_flight.load(Ordering::Acquire) <= self.max_in_flight || server_lost(&self.shared) {
+            return Poll::Ready(());
+        }
+        Poll::Pending
+    }
+
+    fn write(&mut self, data: &[u8]) {
+        self.pending.extend_from_slice(data);
+        if self.pending.len() >= self.submit_bytes {
+            self.submit(self.pending.len() & !63);
+        }
+    }
+
+    fn poll_finish(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<[u8; 16]>> {
+        if !self.finish_sent {
+            self.finish_sent = true;
+            self.submit(self.pending.len());
+            if self.tx.send(Msg::Finish(self.id)).is_err() {
+                self.shared.settle(Err(()));
+            }
+        }
+        let take = |shared: &Shared| shared.outcome.lock().unwrap_or_else(|e| e.into_inner()).take();
+        let mut outcome = take(&self.shared);
+        if outcome.is_none() {
+            self.shared.register(cx);
+            outcome = take(&self.shared);
+        }
+        match outcome {
+            None => Poll::Pending,
+            Some(result) => {
+                self.done = true;
+                Poll::Ready(result.map_err(|()| io::Error::other("md5 lane server stopped before the digest was ready")))
+            }
+        }
+    }
+}
+
+fn server_lost(shared: &Shared) -> bool {
+    matches!(*shared.outcome.lock().unwrap_or_else(|e| e.into_inner()), Some(Err(())))
+}
+
+impl Drop for ServerLane {
+    fn drop(&mut self) {
+        if !self.done {
+            let _ = self.tx.send(Msg::Release(self.id));
+        }
+    }
+}
+
+/// The server-side half of an upload.
+struct Lane<S> {
+    state: S,
+    shared: Arc<Shared>,
+    queue: std::collections::VecDeque<Bytes>,
+    /// The chunk being hashed in the current batch.
+    current: Option<Bytes>,
+    finishing: bool,
+}
+
+struct Worker<E: Md5LaneEngine> {
+    engine: E,
+    rx: Receiver<Msg>,
+    lanes: HashMap<u64, Lane<E::State>>,
+    batch_wait: Duration,
+    alive: Arc<AtomicBool>,
+    stats: Arc<LaneServerStats>,
+}
+
+impl<E: Md5LaneEngine> Drop for Worker<E> {
+    /// Runs on normal exit and on panic alike: nobody may be left waiting for a digest.
+    fn drop(&mut self) {
+        self.alive.store(false, Ordering::Release);
+        for lane in self.lanes.values() {
+            lane.shared.settle(Err(()));
+        }
+    }
+}
+
+impl<E: Md5LaneEngine> Worker<E> {
+    fn run(mut self) {
+        loop {
+            let ready = self.lanes.values().filter(|lane| !lane.queue.is_empty()).count();
+            let first = if ready == 0 {
+                match self.rx.recv() {
+                    Ok(msg) => Some(msg),
+                    Err(_) => return,
+                }
+            } else if ready < self.engine.preferred_batch() && !self.batch_wait.is_zero() {
+                match self.rx.recv_timeout(self.batch_wait) {
+                    Ok(msg) => Some(msg),
+                    Err(RecvTimeoutError::Timeout) => None,
+                    Err(RecvTimeoutError::Disconnected) => return,
+                }
+            } else {
+                None
+            };
+            if let Some(msg) = first {
+                self.accept(msg);
+            }
+            loop {
+                match self.rx.try_recv() {
+                    Ok(msg) => self.accept(msg),
+                    Err(TryRecvError::Empty) => break,
+                    Err(TryRecvError::Disconnected) => return,
+                }
+            }
+            self.hash_one_round();
+            self.finish_drained_lanes();
+        }
+    }
+
+    fn accept(&mut self, msg: Msg) {
+        match msg {
+            Msg::Open(id, shared) => {
+                let lane = Lane {
+                    state: self.engine.start(),
+                    shared,
+                    queue: Default::default(),
+                    current: None,
+                    finishing: false,
+                };
+                self.lanes.insert(id, lane);
+            }
+            Msg::Chunk(id, bytes) => {
+                if let Some(lane) = self.lanes.get_mut(&id) {
+                    lane.queue.push_back(bytes);
+                }
+            }
+            Msg::Finish(id) => {
+                if let Some(lane) = self.lanes.get_mut(&id) {
+                    lane.finishing = true;
+                }
+            }
+            Msg::Release(id) => {
+                self.lanes.remove(&id);
+            }
+        }
+    }
+
+    /// Every lane with queued data contributes its oldest chunk; the engine advances them together.
+    fn hash_one_round(&mut self) {
+        let mut in_batch = 0u64;
+        for lane in self.lanes.values_mut() {
+            lane.current = lane.queue.pop_front();
+            in_batch += u64::from(lane.current.is_some());
+        }
+        if in_batch == 0 {
+            return;
+        }
+        self.engine.update_many(self.lanes.values_mut().filter_map(|lane| {
+            let Lane { state, current, .. } = lane;
+            current.as_deref().map(|chunk| (state, chunk))
+        }));
+        self.stats.batches.fetch_add(1, Ordering::Relaxed);
+        self.stats.batched_lanes.fetch_add(in_batch, Ordering::Relaxed);
+        for lane in self.lanes.values_mut() {
+            if let Some(chunk) = lane.current.take() {
+                lane.shared.in_flight.fetch_sub(chunk.len(), Ordering::AcqRel);
+                lane.shared.wake();
+            }
+        }
+    }
+
+    fn finish_drained_lanes(&mut self) {
+        let done: Vec<u64> = self
+            .lanes
+            .iter()
+            .filter(|(_, lane)| lane.finishing && lane.queue.is_empty())
+            .map(|(id, _)| *id)
+            .collect();
+        for id in done {
+            if let Some(lane) = self.lanes.remove(&id) {
+                lane.shared.settle(Ok(self.engine.finalize(lane.state)));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{BadDigest, EtagReader, EtagResolvable};
+    use md5::{Digest, Md5};
+    use rustfs_utils::hash_lanes::InlineLaneEngine;
+    use std::pin::Pin;
+    use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
+
+    fn md5_hex(data: &[u8]) -> String {
+        hex_simd::encode_to_string(Md5::digest(data), hex_simd::AsciiCase::Lower)
+    }
+
+    fn payload(seed: u64, len: usize) -> Vec<u8> {
+        let mut x = seed | 1;
+        (0..len)
+            .map(|_| {
+                x ^= x << 13;
+                x ^= x >> 7;
+                x ^= x << 17;
+                x as u8
+            })
+            .collect()
+    }
+
+    /// Hands out at most `step` bytes per read, like a network body.
+    struct Dribble {
+        data: Vec<u8>,
+        at: usize,
+        step: usize,
+    }
+
+    impl AsyncRead for Dribble {
+        fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+            let n = self.step.min(buf.remaining()).min(self.data.len() - self.at);
+            let at = self.at;
+            buf.put_slice(&self.data[at..at + n]);
+            self.at += n;
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    fn always_server(config: LaneServerConfig) -> LaneServerConfig {
+        LaneServerConfig {
+            min_streams: 1,
+            ..config
+        }
+    }
+
+    async fn upload_many(server: &Arc<LaneServer>, sizes: &[usize], step: usize) {
+        let mut tasks = Vec::new();
+        for (i, &len) in sizes.iter().enumerate() {
+            let data = payload(i as u64 + 1, len);
+            let want = md5_hex(&data);
+            let mut reader = EtagReader::with_sink(
+                Dribble {
+                    data: data.clone(),
+                    at: 0,
+                    step,
+                },
+                None,
+                server.as_ref(),
+            );
+            tasks.push(tokio::spawn(async move {
+                assert_eq!(reader.try_resolve_etag(), None, "no etag before EOF");
+                let mut out = Vec::new();
+                reader.read_to_end(&mut out).await.expect("read");
+                assert_eq!(out, data, "bytes pass through unchanged");
+                assert_eq!(reader.try_resolve_etag(), Some(want), "stream {i} len {len}");
+            }));
+        }
+        for task in tasks {
+            task.await.expect("upload task");
+        }
+    }
+
+    const SIZES: [usize; 24] = [
+        0,
+        1,
+        55,
+        56,
+        63,
+        64,
+        65,
+        4096,
+        65_535,
+        65_536,
+        262_143,
+        262_144,
+        262_145,
+        300_000,
+        777_777,
+        1_048_576,
+        1_048_577,
+        2_000_003,
+        3,
+        128,
+        500_000,
+        999_999,
+        1_500_000,
+        64 * 1000,
+    ];
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_uploads_match_md5_inline_engine() {
+        let server = LaneServer::start(InlineLaneEngine, always_server(Default::default())).unwrap();
+        upload_many(&server, &SIZES, 40_000).await;
+        assert_eq!(server.stats().lanes_opened.load(Ordering::Relaxed), SIZES.len() as u64);
+        assert_eq!(server.stats().inline_fallbacks.load(Ordering::Relaxed), 0);
+    }
+
+    #[cfg(feature = "hash-md5-simd")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_uploads_match_md5_simd_engine() {
+        let engine = rustfs_utils::hash_lanes::SimdLaneEngine::new();
+        let server = LaneServer::start(engine, always_server(Default::default())).unwrap();
+        upload_many(&server, &SIZES, 40_000).await;
+        upload_many(&server, &[1_048_576; 32], 1 << 20).await;
+        let stats = server.stats();
+        let (batches, lanes) = (stats.batches.load(Ordering::Relaxed), stats.batched_lanes.load(Ordering::Relaxed));
+        assert!(lanes > batches, "uploads were never hashed together: {lanes} lanes in {batches} batches");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn tiny_limits_exercise_back_pressure() {
+        let config = LaneServerConfig {
+            submit_bytes: 64,
+            max_in_flight_bytes: 128,
+            batch_wait: Duration::ZERO,
+            min_streams: 1,
+            threads: 3,
+        };
+        let server = LaneServer::start(InlineLaneEngine, config).unwrap();
+        upload_many(&server, &[0, 63, 64, 65, 10_000, 100_001, 250_000], 997).await;
+    }
+
+    #[tokio::test]
+    async fn content_md5_is_verified_at_eof() {
+        let server = LaneServer::start(InlineLaneEngine, always_server(Default::default())).unwrap();
+        let data = payload(9, 300_000);
+        let good = md5_hex(&data);
+
+        let mut ok = EtagReader::with_sink(&data[..], Some(good.clone()), server.as_ref());
+        ok.read_to_end(&mut Vec::new()).await.expect("matching Content-MD5");
+        assert_eq!(ok.get_etag(), Some(good.clone()));
+
+        let wrong = "0".repeat(32);
+        let mut bad = EtagReader::with_sink(&data[..], Some(wrong.clone()), server.as_ref());
+        let err = bad.read_to_end(&mut Vec::new()).await.expect_err("mismatching Content-MD5");
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+        let digest = err.get_ref().and_then(|e| e.downcast_ref::<BadDigest>()).expect("BadDigest");
+        assert_eq!(
+            (digest.expected_md5.as_str(), digest.calculated_md5.as_str()),
+            (wrong.as_str(), good.as_str())
+        );
+    }
+
+    #[tokio::test]
+    async fn few_uploads_stay_inline_and_are_counted_while_alive() {
+        let server = LaneServer::start(
+            InlineLaneEngine,
+            LaneServerConfig {
+                min_streams: 3,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+        let data = payload(4, 100_000);
+        let first = EtagReader::with_sink(&data[..], None, server.as_ref());
+        let second = EtagReader::with_sink(&data[..], None, server.as_ref());
+        let mut third = EtagReader::with_sink(&data[..], None, server.as_ref());
+        let stats = server.stats();
+        assert_eq!(
+            (stats.inline_fallbacks.load(Ordering::Relaxed), stats.lanes_opened.load(Ordering::Relaxed)),
+            (2, 1)
+        );
+        third.read_to_end(&mut Vec::new()).await.unwrap();
+        assert_eq!(third.get_etag(), Some(md5_hex(&data)));
+        drop((first, second, third));
+        let mut alone = EtagReader::with_sink(&data[..], None, server.as_ref());
+        assert_eq!(
+            stats.inline_fallbacks.load(Ordering::Relaxed),
+            3,
+            "the count dropped when the readers did"
+        );
+        alone.read_to_end(&mut Vec::new()).await.unwrap();
+        assert_eq!(alone.get_etag(), Some(md5_hex(&data)));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn abandoned_uploads_do_not_disturb_the_rest() {
+        let server = LaneServer::start(InlineLaneEngine, always_server(Default::default())).unwrap();
+        for round in 0..20 {
+            let data = payload(round, 600_000);
+            let mut abandoned = EtagReader::with_sink(
+                Dribble {
+                    data: data.clone(),
+                    at: 0,
+                    step: 70_000,
+                },
+                None,
+                server.as_ref(),
+            );
+            let mut partial = vec![0u8; 300_000];
+            abandoned.read_exact(&mut partial).await.unwrap();
+            drop(abandoned);
+        }
+        upload_many(&server, &[500_000; 8], 50_000).await;
+    }
+
+    /// An engine that dies on its first batch.
+    #[derive(Clone)]
+    struct PanickingEngine;
+
+    impl Md5LaneEngine for PanickingEngine {
+        type State = ();
+        fn name(&self) -> &'static str {
+            "panics"
+        }
+        fn preferred_batch(&self) -> usize {
+            1
+        }
+        fn start(&self) {}
+        fn update_many<'a, I>(&self, _streams: I)
+        where
+            I: IntoIterator<Item = (&'a mut (), &'a [u8])>,
+        {
+            panic!("engine failure injected by test");
+        }
+        fn finalize(&self, _state: ()) -> [u8; 16] {
+            [0; 16]
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn a_dead_server_fails_in_flight_uploads_and_later_ones_go_inline() {
+        let server = LaneServer::start(PanickingEngine, always_server(Default::default())).unwrap();
+        let data = payload(5, 400_000);
+        let mut doomed = EtagReader::with_sink(&data[..], None, server.as_ref());
+        let err = tokio::time::timeout(Duration::from_secs(10), doomed.read_to_end(&mut Vec::new()))
+            .await
+            .expect("a dead server must not leave the upload hanging")
+            .expect_err("no digest can be produced");
+        assert_eq!(err.kind(), io::ErrorKind::Other);
+        assert_eq!(doomed.try_resolve_etag(), None);
+
+        let mut later = EtagReader::with_sink(&data[..], None, server.as_ref());
+        later.read_to_end(&mut Vec::new()).await.expect("inline fallback");
+        assert_eq!(later.get_etag(), Some(md5_hex(&data)));
+    }
+
+    /// Aggregate ETag throughput of N concurrent uploads read in 64 KiB pieces, inline vs the lane
+    /// server. `cargo test -p rustfs-rio --release --features hash-md5-simd -- --ignored --nocapture etag_throughput`
+    #[cfg(feature = "hash-md5-simd")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "benchmark, not a correctness test"]
+    async fn etag_throughput() {
+        const PER_UPLOAD: usize = 64 << 20;
+        let data = Arc::new(payload(7, PER_UPLOAD));
+        let want = md5_hex(&data);
+        for threads in [1usize, 2] {
+            let engine = rustfs_utils::hash_lanes::SimdLaneEngine::new();
+            let server = LaneServer::start(
+                engine,
+                LaneServerConfig {
+                    threads,
+                    ..always_server(Default::default())
+                },
+            )
+            .unwrap();
+            println!("--- lane server threads = {threads}");
+            println!(
+                "engine={} tokio workers=4, {} MiB per upload, 64 KiB reads",
+                server.engine_name(),
+                PER_UPLOAD >> 20
+            );
+            println!(
+                "{:>7} | {:>12} {:>12} | {:>12} {:>12} | {:>6} {:>8} {:>11}",
+                "uploads", "inline MiB/s", "cpu-s / GiB", "server MiB/s", "cpu-s / GiB", "speed", "cpu cost", "lanes/batch"
+            );
+            // User + system CPU time of the whole process (all threads), in seconds.
+            fn cpu_seconds() -> f64 {
+                let t = rustix::time::clock_gettime(rustix::time::ClockId::ProcessCPUTime);
+                t.tv_sec as f64 + t.tv_nsec as f64 / 1e9
+            }
+            for n in [1usize, 2, 3, 4, 8, 16, 32] {
+                let mut rates = [0f64; 2];
+                let mut cpu = [0f64; 2];
+                let before = (
+                    server.stats().batches.load(Ordering::Relaxed),
+                    server.stats().batched_lanes.load(Ordering::Relaxed),
+                );
+                for (slot, sink) in [
+                    Arc::new(InlineSink) as Arc<dyn Md5StreamSink>,
+                    server.clone() as Arc<dyn Md5StreamSink>,
+                ]
+                .into_iter()
+                .enumerate()
+                {
+                    let mut best = 0f64;
+                    let cpu_before = cpu_seconds();
+                    for _ in 0..3 {
+                        let started = std::time::Instant::now();
+                        let tasks: Vec<_> = (0..n)
+                            .map(|_| {
+                                let (data, sink, want) = (data.clone(), sink.clone(), want.clone());
+                                tokio::spawn(async move {
+                                    let source = Dribble {
+                                        data: data.to_vec(),
+                                        at: 0,
+                                        step: 64 * 1024,
+                                    };
+                                    let mut reader = EtagReader::with_sink(source, None, sink.as_ref());
+                                    let mut buf = vec![0u8; 64 * 1024];
+                                    while reader.read(&mut buf).await.unwrap() != 0 {}
+                                    assert_eq!(reader.get_etag(), Some(want));
+                                })
+                            })
+                            .collect();
+                        for task in tasks {
+                            task.await.unwrap();
+                        }
+                        best = best.max((n * PER_UPLOAD) as f64 / started.elapsed().as_secs_f64() / 1048576.0);
+                    }
+                    rates[slot] = best;
+                    cpu[slot] = (cpu_seconds() - cpu_before) / (3.0 * (n * PER_UPLOAD) as f64 / (1u64 << 30) as f64);
+                }
+                let batches = server.stats().batches.load(Ordering::Relaxed) - before.0;
+                let lanes = server.stats().batched_lanes.load(Ordering::Relaxed) - before.1;
+                println!(
+                    "{n:>7} | {:>12.0} {:>12.2} | {:>12.0} {:>12.2} | {:>5.2}x {:>7.2}x {:>11.1}",
+                    rates[0],
+                    cpu[0],
+                    rates[1],
+                    cpu[1],
+                    rates[1] / rates[0],
+                    cpu[1] / cpu[0],
+                    lanes as f64 / batches.max(1) as f64
+                );
+            }
+        }
+    }
+}

crates/s3-client/Cargo.toml

@@ -46,7 +46,6 @@ hyper = { workspace = true, features = ["http2", "http1", "server"] }
 hyper-rustls = { workspace = true, default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs"] }
 hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-graceful", "tracing"] }
 lazy_static.workspace = true
-md-5.workspace = true
 quick-xml = { workspace = true, features = ["serialize", "async-tokio"] }
 rand = { workspace = true, features = ["serde"] }
 rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }

crates/s3-client/src/transition_api.rs

@@ -46,8 +46,6 @@ use hyper::body::Bytes;
 use hyper::body::Incoming;
 use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
 use hyper_util::{client::legacy::Client, client::legacy::connect::HttpConnector, rt::TokioExecutor};
-use md5::Digest;
-use md5::Md5;
 use rand::{Rng, RngExt};
 use rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE;
 use rustfs_rio::HashReader;

crates/utils/Cargo.toml

@@ -41,6 +41,8 @@ hyper = { workspace = true, optional = true, features = ["http2", "http1", "serv
 local-ip-address = { workspace = true, optional = true }
 lz4 = { workspace = true, optional = true }
 md-5 = { workspace = true, optional = true }
+fast-md5 = { workspace = true, optional = true }
+md5-simd = { workspace = true, optional = true }
 netif = { workspace = true, optional = true }
 regex = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true }
@@ -89,6 +91,12 @@ compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
 string = ["dep:regex"]
 crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1", "dep:sha2"]
 hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake2", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:crc-fast"]
+# Swap the MD5 backend behind `hash_stream::Md5Stream` to `fast-md5` (hand-written x86_64/aarch64 kernels).
+# Deliberately outside `default` and `full`: enabling it is an explicit build decision. `md-5` stays
+# compiled in as the differential-test reference.
+hash-md5-fast-md5 = ["hash", "dep:fast-md5"]
+# Multi-stream MD5 kernel behind `hash_lanes::Md5LaneEngine` (Port C). Trial wiring.
+hash-md5-simd = ["hash", "dep:md5-simd"]
 os = ["dep:rustix", "dep:tempfile", "dep:windows"]  # operating system utilities
 integration = []  # integration test features
 http = ["dep:convert_case", "dep:http", "dep:regex"]

crates/utils/benches/hash_hotpath_benchmark.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
-use rustfs_utils::HashAlgorithm;
+use rustfs_utils::{HashAlgorithm, Md5Stream};
 use std::hint::black_box;
 
 fn generate_payload(size: usize) -> Vec<u8> {
@@ -47,6 +47,21 @@ fn bench_hash_hotpaths(c: &mut Criterion) {
                 });
             });
         }
+        // The ETag path never sees the object in one piece: EtagReader is fed
+        // whatever the body stream yields. Model that with 256 KiB updates.
+        group.bench_with_input(
+            BenchmarkId::new("md5_stream_256KiB_chunks", payload_name),
+            payload.as_slice(),
+            |b, payload| {
+                b.iter(|| {
+                    let mut hasher = Md5Stream::new();
+                    for chunk in black_box(payload).chunks(256 * 1024) {
+                        hasher.update(chunk);
+                    }
+                    black_box(hasher.finalize()[0]);
+                });
+            },
+        );
     }
     group.finish();
 }

crates/utils/src/hash.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub use crate::hash_stream::Md5Stream;
 use blake2::Blake2b512;
 use highway::{HighwayHash, HighwayHasher, Key};
-use md5::Md5;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 
@@ -113,7 +113,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(Md5Stream::digest(data)),
             HashAlgorithm::HighwayHash256 => {
                 let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
                 hasher.append(data);
@@ -148,11 +148,11 @@ impl HashAlgorithm {
     {
         match self {
             HashAlgorithm::Md5 => {
-                let mut hasher = Md5::new();
+                let mut hasher = Md5Stream::new();
                 for slice in slices {
                     hasher.update(slice);
                 }
-                HashEncoded::Md5(hasher.finalize().into())
+                HashEncoded::Md5(hasher.finalize())
             }
             HashAlgorithm::HighwayHash256 => {
                 let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
@@ -302,16 +302,41 @@ mod tests {
 
     #[test]
     fn test_hash_encode_md5() {
+        // Same input and digest that crates/checksums and crates/rio pin, so a
+        // backend change on any side fails loudly here too.
         let data = b"test data";
         let hash = HashAlgorithm::Md5.hash_encode(data);
         let hash = hash.as_ref();
         assert_eq!(hash.len(), 16);
+        assert_eq!(
+            hex_simd::encode_to_string(hash, hex_simd::AsciiCase::Lower),
+            "eb733a00c0c9d336e65691a37ab54293"
+        );
         // MD5 should be deterministic
         let hash2 = HashAlgorithm::Md5.hash_encode(data);
         let hash2 = hash2.as_ref();
         assert_eq!(hash, hash2);
     }
 
+    /// RFC 1321 vectors through the one-shot interface: `HashAlgorithm::Md5`
+    /// and `Md5Stream` must be the same MD5 whatever backend is compiled in.
+    #[test]
+    fn test_hash_encode_md5_rfc1321_vectors() {
+        for (input, want) in [
+            (&b""[..], "d41d8cd98f00b204e9800998ecf8427e"),
+            (&b"abc"[..], "900150983cd24fb0d6963f7d28e17f72"),
+            (&b"message digest"[..], "f96b697d7cb7938d525a2f31aaf161d0"),
+            (
+                &b"12345678901234567890123456789012345678901234567890123456789012345678901234567890"[..],
+                "57edf4a22be3c955ac49da2e2107b67a",
+            ),
+        ] {
+            let got = HashAlgorithm::Md5.hash_encode(input);
+            assert_eq!(hex_simd::encode_to_string(got.as_ref(), hex_simd::AsciiCase::Lower), want);
+            assert_eq!(got.as_ref(), Md5Stream::digest(input));
+        }
+    }
+
     #[test]
     fn test_hash_encode_highway() {
         let data = b"test data";

crates/utils/src/hash_lanes.rs

@@ -0,0 +1,307 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Multi-stream MD5 port (Port C).
+//!
+//! [`hash_stream::Md5Stream`](crate::hash_stream::Md5Stream) is the one-stream-at-a-time
+//! interface. This module is its many-streams-at-a-time sibling: a component that holds many
+//! in-flight MD5 streams (the ETag lane server) advances them together through
+//! [`Md5LaneEngine`] and never names a concrete kernel.
+//!
+//! The trait is the intersection of what any engine can offer, same rule as `Md5Stream`:
+//! an opaque per-stream state, "feed these slices to these states", and a consuming
+//! `finalize`. It asks for no `Clone`/`Copy`, no contiguous state array, no alignment and no
+//! equal lengths, so [`InlineLaneEngine`] (a plain loop over `Md5Stream`) is a complete
+//! implementation and the always-available fallback.
+//!
+//! | engine | cargo feature | what it does |
+//! |---|---|---|
+//! | [`InlineLaneEngine`] | `hash` | each stream through `Md5Stream`, no batching |
+//! | [`SimdLaneEngine`]   | `hash-md5-simd` | `md5-simd` incremental `update_many`: 4/8/16 streams per SIMD register |
+
+use crate::hash_stream::Md5Stream;
+
+/// Advances many independent MD5 streams at once.
+///
+/// Callers own the states and may keep them anywhere (a slab, a map, a per-connection struct);
+/// the engine only borrows them for the duration of one [`update_many`](Self::update_many).
+pub trait Md5LaneEngine: Send + Sync + 'static {
+    /// One in-flight stream. Opaque to the caller.
+    type State: Send + 'static;
+
+    /// Diagnostic name (startup logs, metrics labels); never part of any format.
+    fn name(&self) -> &'static str;
+
+    /// How many streams make a batch that runs at peak throughput. A scheduler gathers up to
+    /// this many before calling [`update_many`](Self::update_many); any other count is still
+    /// correct and never slower than feeding the streams one by one.
+    fn preferred_batch(&self) -> usize;
+
+    /// A new, empty stream.
+    fn start(&self) -> Self::State;
+
+    /// Append each slice to its stream. Slices may differ in length and may be empty.
+    fn update_many<'a, I>(&self, streams: I)
+    where
+        I: IntoIterator<Item = (&'a mut Self::State, &'a [u8])>;
+
+    /// Consume a stream and return its digest.
+    fn finalize(&self, state: Self::State) -> [u8; 16];
+}
+
+/// No batching: every stream goes through [`Md5Stream`]. Behaviourally identical to hashing in
+/// place, which makes it both the fallback and the reference for differential tests.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct InlineLaneEngine;
+
+impl Md5LaneEngine for InlineLaneEngine {
+    type State = Md5Stream;
+
+    fn name(&self) -> &'static str {
+        Md5Stream::ACTIVE_BACKEND
+    }
+
+    fn preferred_batch(&self) -> usize {
+        1
+    }
+
+    fn start(&self) -> Md5Stream {
+        Md5Stream::new()
+    }
+
+    fn update_many<'a, I>(&self, streams: I)
+    where
+        I: IntoIterator<Item = (&'a mut Md5Stream, &'a [u8])>,
+    {
+        for (state, data) in streams {
+            state.update(data);
+        }
+    }
+
+    fn finalize(&self, state: Md5Stream) -> [u8; 16] {
+        state.finalize()
+    }
+}
+
+#[cfg(feature = "hash-md5-simd")]
+pub use simd::SimdLaneEngine;
+
+#[cfg(feature = "hash-md5-simd")]
+mod simd {
+    use super::Md5LaneEngine;
+    use md5_simd::{Md5Engine, Md5State};
+
+    /// `md5-simd` behind the port. The backing crate's types never leave this module.
+    #[derive(Clone, Copy, Debug, Default)]
+    pub struct SimdLaneEngine(Md5Engine);
+
+    /// Opaque stream state of [`SimdLaneEngine`].
+    pub struct SimdLaneState(Md5State);
+
+    /// `Md5Engine::update_many` wants contiguous states; the port hands out scattered ones.
+    /// States are small and `Copy`, so each window is copied in, advanced, and copied back.
+    const WINDOW: usize = 64;
+
+    impl SimdLaneEngine {
+        /// The best engine for the running CPU.
+        pub fn new() -> Self {
+            Self(Md5Engine::new())
+        }
+
+        fn flush(&self, slots: &mut [Option<&mut SimdLaneState>], inputs: &[&[u8]]) {
+            let mut states = [Md5State::new(); WINDOW];
+            for (state, slot) in states.iter_mut().zip(slots.iter()) {
+                if let Some(slot) = slot {
+                    *state = slot.0;
+                }
+            }
+            self.0.update_many(&mut states[..slots.len()], inputs);
+            for (state, slot) in states.iter().zip(slots.iter_mut()) {
+                if let Some(slot) = slot.take() {
+                    slot.0 = *state;
+                }
+            }
+        }
+    }
+
+    impl Md5LaneEngine for SimdLaneEngine {
+        type State = SimdLaneState;
+
+        fn name(&self) -> &'static str {
+            self.0.simd_name()
+        }
+
+        fn preferred_batch(&self) -> usize {
+            self.0.lanes().max(1)
+        }
+
+        fn start(&self) -> SimdLaneState {
+            SimdLaneState(Md5State::new())
+        }
+
+        fn update_many<'a, I>(&self, streams: I)
+        where
+            I: IntoIterator<Item = (&'a mut SimdLaneState, &'a [u8])>,
+        {
+            let mut slots: [Option<&'a mut SimdLaneState>; WINDOW] = [const { None }; WINDOW];
+            let mut inputs: [&[u8]; WINDOW] = [&[]; WINDOW];
+            let mut n = 0;
+            for (state, data) in streams {
+                slots[n] = Some(state);
+                inputs[n] = data;
+                n += 1;
+                if n == WINDOW {
+                    self.flush(&mut slots, &inputs);
+                    n = 0;
+                }
+            }
+            self.flush(&mut slots[..n], &inputs[..n]);
+        }
+
+        fn finalize(&self, state: SimdLaneState) -> [u8; 16] {
+            state.0.finalize()
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    struct Rng(u64);
+    impl Rng {
+        fn next(&mut self) -> u64 {
+            self.0 ^= self.0 << 13;
+            self.0 ^= self.0 >> 7;
+            self.0 ^= self.0 << 17;
+            self.0
+        }
+        fn bytes(&mut self, len: usize) -> Vec<u8> {
+            (0..len).map(|_| self.next() as u8).collect()
+        }
+    }
+
+    /// The shape the lane server has: streams live in a map keyed by id, arrive and finish at
+    /// different times, get a block-aligned chunk most ticks, nothing on some ticks, and an odd
+    /// tail at the end. Every digest must equal the one-stream interface.
+    fn lane_server_shape<E: Md5LaneEngine>(engine: &E) {
+        let mut rng = Rng(0x5eed_1234_abcd_ef01);
+        let objects: Vec<Vec<u8>> = (0..45)
+            .map(|i| {
+                let len = match i % 5 {
+                    0 => 0,
+                    1 => 777,
+                    _ => 300_000 + (rng.next() % 200_000) as usize,
+                };
+                rng.bytes(len)
+            })
+            .collect();
+        let mut in_flight: HashMap<usize, (E::State, usize)> = HashMap::new();
+        let mut digests: HashMap<usize, [u8; 16]> = HashMap::new();
+        let mut next_object = 0;
+        for tick in 0.. {
+            // admit a few new streams per tick
+            for _ in 0..(1 + rng.next() % 6) {
+                if next_object < objects.len() {
+                    in_flight.insert(next_object, (engine.start(), 0));
+                    next_object += 1;
+                }
+            }
+            if in_flight.is_empty() {
+                break;
+            }
+            // this tick's chunk per stream: 64 KiB, or nothing for a stalled client
+            let plan: HashMap<usize, (usize, usize)> = in_flight
+                .iter()
+                .map(|(&id, &(_, off))| {
+                    let stalled = (id + tick) % 7 == 0;
+                    let take = if stalled { 0 } else { (objects[id].len() - off).min(65536) };
+                    (id, (off, take))
+                })
+                .collect();
+            engine.update_many(in_flight.iter_mut().map(|(id, (state, _))| {
+                let (off, take) = plan[id];
+                (state, &objects[*id][off..off + take])
+            }));
+            for (id, (off, take)) in plan {
+                in_flight.get_mut(&id).unwrap().1 = off + take;
+                if off + take == objects[id].len() && take < 65536 && !((id + tick) % 7 == 0 && !objects[id].is_empty()) {
+                    let (state, _) = in_flight.remove(&id).unwrap();
+                    digests.insert(id, engine.finalize(state));
+                }
+            }
+        }
+        assert_eq!(digests.len(), objects.len());
+        for (id, object) in objects.iter().enumerate() {
+            assert_eq!(
+                digests[&id],
+                Md5Stream::digest(object),
+                "engine {} object {id} len {}",
+                engine.name(),
+                object.len()
+            );
+        }
+    }
+
+    #[test]
+    fn inline_engine_matches_md5_stream() {
+        assert_eq!(InlineLaneEngine.preferred_batch(), 1);
+        lane_server_shape(&InlineLaneEngine);
+    }
+
+    #[cfg(feature = "hash-md5-simd")]
+    #[test]
+    fn simd_engine_matches_md5_stream() {
+        let engine = SimdLaneEngine::new();
+        assert!(engine.preferred_batch() >= 1);
+        assert!(!engine.name().is_empty());
+        lane_server_shape(&engine);
+    }
+
+    /// Throughput through the port, 32 streams x 1 MiB chunks. `cargo test -p rustfs-utils
+    /// --release --features hash-md5-simd -- --ignored --nocapture lanes_throughput`.
+    #[test]
+    #[ignore = "benchmark, not a correctness test"]
+    fn lanes_throughput() {
+        fn run<E: Md5LaneEngine>(engine: &E, data: &[Vec<u8>]) -> f64 {
+            let mut best = 0f64;
+            for _ in 0..3 {
+                let mut states: Vec<E::State> = data.iter().map(|_| engine.start()).collect();
+                let t = std::time::Instant::now();
+                for off in (0..data[0].len()).step_by(1 << 20) {
+                    engine.update_many(states.iter_mut().zip(data).map(|(s, d)| (s, &d[off..off + (1 << 20)])));
+                }
+                let digests: Vec<[u8; 16]> = states.into_iter().map(|s| engine.finalize(s)).collect();
+                let secs = t.elapsed().as_secs_f64();
+                assert_eq!(digests[7], Md5Stream::digest(&data[7]));
+                best = best.max((data.len() * data[0].len()) as f64 / secs / 1048576.0);
+            }
+            best
+        }
+        let mut rng = Rng(42);
+        let data: Vec<Vec<u8>> = (0..32).map(|_| rng.bytes(8 << 20)).collect();
+        println!("{:<20} {:>7.0} MiB/s", InlineLaneEngine.name(), run(&InlineLaneEngine, &data));
+        #[cfg(feature = "hash-md5-simd")]
+        {
+            let engine = SimdLaneEngine::new();
+            println!(
+                "md5-simd/{:<11} {:>7.0} MiB/s (batch {})",
+                engine.name(),
+                run(&engine, &data),
+                engine.preferred_batch()
+            );
+        }
+    }
+}

crates/utils/src/hash_stream.rs

@@ -0,0 +1,389 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Streaming (incremental) hash interface.
+//!
+//! [`hash`](crate::hash) owns the one-shot interface (`HashAlgorithm::hash_encode`).
+//! This module owns the streaming shape: feed bytes over time, take the digest
+//! at the end. Both are the *only* places RustFS touches a concrete MD5
+//! implementation; every other crate goes through [`Md5Stream`] or
+//! `HashAlgorithm::Md5`, never through the backing crate directly.
+//!
+//! Backend selection lives in the private [`backend`] module. Exactly one
+//! backend is compiled in:
+//!
+//! | cargo feature (on `rustfs-utils`) | backend |
+//! |---|---|
+//! | `hash` (default for consumers)    | RustCrypto `md-5`, portable Rust |
+//! | `hash-md5-fast-md5`               | `fast-md5`, hand-written x86_64 / aarch64 kernels |
+//!
+//! `rustfs-rio` and the `rustfs` binary forward `hash-md5-fast-md5`, so
+//! `cargo build --features hash-md5-fast-md5` switches the whole tree. Adding
+//! another backend means adding one `Md5Inner` behind a cargo feature in
+//! [`backend`] and nothing else in the tree changes.
+//!
+//! # Why `finalize` consumes `self` and there is no `Clone`
+//!
+//! Not every MD5 implementation can snapshot its state (an assembly backend may
+//! keep an uninitialised block buffer, for example). The interface is therefore
+//! the intersection of what any backend can offer: append bytes, consume the
+//! hasher for the digest, or take the digest and start over. Code that needs a
+//! digest "so far" while continuing to feed bytes must restructure, not clone;
+//! keeping that constraint on the default backend means switching backends can
+//! never turn into a compile error somewhere else in the tree.
+
+use std::fmt;
+use std::io;
+
+/// Incremental MD5 hasher: the single streaming MD5 entry point for RustFS.
+///
+/// Used for the S3 ETag of every PUT / UploadPart (`rustfs_rio::EtagReader`),
+/// for `Content-MD5` verification, and for the MD5 additional checksum.
+/// The digest is plain MD5 (RFC 1321); this type never changes what bytes
+/// mean, only which implementation computes them.
+pub struct Md5Stream(backend::Md5Inner);
+
+impl Md5Stream {
+    /// Digest length in bytes.
+    pub const OUTPUT_SIZE: usize = 16;
+
+    /// Name of the backend compiled into this build. Diagnostic only
+    /// (startup logs, tests); never part of any on-wire or on-disk format.
+    pub const ACTIVE_BACKEND: &'static str = backend::NAME;
+
+    /// Create an empty hasher.
+    #[inline]
+    pub fn new() -> Self {
+        Self(backend::Md5Inner::new())
+    }
+
+    /// Append bytes. Any slice length is accepted; splitting the input across
+    /// calls never changes the digest.
+    #[inline]
+    pub fn update(&mut self, data: &[u8]) {
+        self.0.update(data);
+    }
+
+    /// Consume the hasher and return the digest.
+    #[inline]
+    pub fn finalize(self) -> [u8; Self::OUTPUT_SIZE] {
+        self.0.finalize()
+    }
+
+    /// Return the digest of everything fed so far and reset this hasher to the
+    /// empty state, as if freshly constructed.
+    #[inline]
+    pub fn finalize_reset(&mut self) -> [u8; Self::OUTPUT_SIZE] {
+        std::mem::replace(&mut self.0, backend::Md5Inner::new()).finalize()
+    }
+
+    /// One-shot digest of `data`.
+    #[inline]
+    pub fn digest(data: &[u8]) -> [u8; Self::OUTPUT_SIZE] {
+        let mut hasher = Self::new();
+        hasher.update(data);
+        hasher.finalize()
+    }
+}
+
+impl Default for Md5Stream {
+    #[inline]
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl fmt::Debug for Md5Stream {
+    /// Deliberately opaque: never print internal state or partial digests.
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("Md5Stream")
+    }
+}
+
+impl io::Write for Md5Stream {
+    #[inline]
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.update(buf);
+        Ok(buf.len())
+    }
+
+    #[inline]
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+/// Concrete MD5 backends. Exactly one `Md5Inner` is compiled in.
+///
+/// Every backend exposes the same three inherent methods
+/// (`new` / `update` / `finalize(self)`); [`Md5Stream`] is written against
+/// that shape only, so a backend does not need `Clone`, `Default`, or any
+/// trait from the backing crate.
+mod backend {
+    #[cfg(not(feature = "hash-md5-fast-md5"))]
+    pub(super) const NAME: &str = "md-5";
+    #[cfg(feature = "hash-md5-fast-md5")]
+    pub(super) const NAME: &str = "fast-md5";
+
+    /// RustCrypto `md-5`: the default backend.
+    #[cfg(not(feature = "hash-md5-fast-md5"))]
+    pub(super) struct Md5Inner(md5::Md5);
+
+    #[cfg(not(feature = "hash-md5-fast-md5"))]
+    impl Md5Inner {
+        #[inline]
+        pub(super) fn new() -> Self {
+            use md5::Digest as _;
+            Self(md5::Md5::new())
+        }
+
+        #[inline]
+        pub(super) fn update(&mut self, data: &[u8]) {
+            use md5::Digest as _;
+            self.0.update(data);
+        }
+
+        #[inline]
+        pub(super) fn finalize(self) -> [u8; 16] {
+            use md5::Digest as _;
+            self.0.finalize().into()
+        }
+    }
+
+    /// `fast-md5`: hand-written assembly compression on x86_64 and aarch64,
+    /// portable Rust elsewhere. Selected at compile time (no runtime CPU
+    /// detection); the x86_64 kernel uses baseline integer instructions only,
+    /// so it runs on any x86_64 CPU.
+    #[cfg(feature = "hash-md5-fast-md5")]
+    pub(super) struct Md5Inner(fast_md5::Md5);
+
+    #[cfg(feature = "hash-md5-fast-md5")]
+    impl Md5Inner {
+        #[inline]
+        pub(super) fn new() -> Self {
+            Self(fast_md5::Md5::new())
+        }
+
+        #[inline]
+        pub(super) fn update(&mut self, data: &[u8]) {
+            self.0.update(data);
+        }
+
+        #[inline]
+        pub(super) fn finalize(self) -> [u8; 16] {
+            self.0.finalize()
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write as _;
+
+    fn hex(digest: &[u8]) -> String {
+        hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower)
+    }
+
+    /// RFC 1321 appendix A.5 test suite.
+    const RFC1321_VECTORS: [(&[u8], &str); 7] = [
+        (b"", "d41d8cd98f00b204e9800998ecf8427e"),
+        (b"a", "0cc175b9c0f1b6a831c399e269772661"),
+        (b"abc", "900150983cd24fb0d6963f7d28e17f72"),
+        (b"message digest", "f96b697d7cb7938d525a2f31aaf161d0"),
+        (b"abcdefghijklmnopqrstuvwxyz", "c3fcd3d76192e4007dfb496cca67e13b"),
+        (
+            b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789",
+            "d174ab98d277d9f5a5611c2c9f419d9f",
+        ),
+        (
+            b"12345678901234567890123456789012345678901234567890123456789012345678901234567890",
+            "57edf4a22be3c955ac49da2e2107b67a",
+        ),
+    ];
+
+    /// Deterministic pseudo-random payload; no `rand` dependency needed.
+    fn payload(len: usize, seed: u64) -> Vec<u8> {
+        let mut x = seed | 1;
+        (0..len)
+            .map(|_| {
+                x ^= x << 13;
+                x ^= x >> 7;
+                x ^= x << 17;
+                (x >> 24) as u8
+            })
+            .collect()
+    }
+
+    #[test]
+    fn rfc1321_known_answers_one_shot() {
+        for (input, want) in RFC1321_VECTORS {
+            assert_eq!(hex(&Md5Stream::digest(input)), want, "input {:?}", String::from_utf8_lossy(input));
+        }
+    }
+
+    #[test]
+    fn rfc1321_known_answers_streaming_byte_at_a_time() {
+        for (input, want) in RFC1321_VECTORS {
+            let mut hasher = Md5Stream::new();
+            for byte in input {
+                hasher.update(std::slice::from_ref(byte));
+            }
+            assert_eq!(hex(&hasher.finalize()), want);
+        }
+    }
+
+    #[test]
+    fn chunked_updates_match_one_shot_across_block_boundaries() {
+        // MD5 works on 64-byte blocks with a 56-byte padding threshold; cover
+        // every split near those edges plus splits inside a multi-megabyte body.
+        let small = payload(300, 7);
+        let want = Md5Stream::digest(&small);
+        for split in [0usize, 1, 2, 55, 56, 57, 63, 64, 65, 119, 120, 127, 128, 129, 200, 299, 300] {
+            let mut hasher = Md5Stream::new();
+            hasher.update(&small[..split]);
+            hasher.update(&small[split..]);
+            assert_eq!(hasher.finalize(), want, "split at {split}");
+        }
+
+        let large = payload((1 << 20) + 37, 11);
+        let want = Md5Stream::digest(&large);
+        let mut hasher = Md5Stream::new();
+        let mut offset = 0usize;
+        let mut chunk = 1usize;
+        while offset < large.len() {
+            let end = (offset + chunk).min(large.len());
+            hasher.update(&large[offset..end]);
+            offset = end;
+            chunk = (chunk * 3 + 1) % 200_003 + 1;
+        }
+        assert_eq!(hasher.finalize(), want);
+    }
+
+    #[test]
+    fn finalize_reset_yields_digest_and_restarts_from_empty() {
+        let mut hasher = Md5Stream::new();
+        hasher.update(b"message digest");
+        assert_eq!(hex(&hasher.finalize_reset()), "f96b697d7cb7938d525a2f31aaf161d0");
+        // After the reset the hasher must behave like a fresh one...
+        assert_eq!(hex(&hasher.finalize_reset()), "d41d8cd98f00b204e9800998ecf8427e");
+        // ...including for a new message that spans a block boundary.
+        let body = payload(1000, 3);
+        hasher.update(&body[..500]);
+        hasher.update(&body[500..]);
+        assert_eq!(hasher.finalize(), Md5Stream::digest(&body));
+    }
+
+    #[test]
+    fn write_impl_feeds_the_hasher() {
+        let body = payload(70_000, 5);
+        let mut hasher = Md5Stream::default();
+        hasher.write_all(&body[..1234]).expect("in-memory write cannot fail");
+        hasher.write_all(&body[1234..]).expect("in-memory write cannot fail");
+        hasher.flush().expect("flush is a no-op");
+        assert_eq!(hasher.finalize(), Md5Stream::digest(&body));
+    }
+
+    #[test]
+    fn debug_output_is_opaque() {
+        let mut hasher = Md5Stream::new();
+        hasher.update(b"secret payload");
+        assert_eq!(format!("{hasher:?}"), "Md5Stream");
+    }
+
+    /// Pins which backend this build compiled in, so the `hash-md5-fast-md5`
+    /// CI leg proves the feature actually took effect rather than silently
+    /// testing the default backend twice.
+    #[test]
+    fn active_backend_matches_cargo_feature() {
+        assert_eq!(
+            Md5Stream::ACTIVE_BACKEND,
+            if cfg!(feature = "hash-md5-fast-md5") {
+                "fast-md5"
+            } else {
+                "md-5"
+            }
+        );
+    }
+
+    #[test]
+    fn output_size_matches_md5() {
+        assert_eq!(Md5Stream::OUTPUT_SIZE, 16);
+        assert_eq!(Md5Stream::digest(b"").len(), Md5Stream::OUTPUT_SIZE);
+    }
+
+    #[test]
+    fn finalize_reset_sequence_matches_independent_digests() {
+        // One hasher reused for several messages must produce exactly the
+        // digests of several fresh hashers; nothing may leak between messages.
+        let messages = [payload(0, 1), payload(1, 2), payload(63, 3), payload(64, 4), payload(4097, 5)];
+        let mut hasher = Md5Stream::new();
+        for message in &messages {
+            let (head, tail) = message.split_at(message.len() / 2);
+            hasher.update(head);
+            hasher.update(tail);
+            assert_eq!(hasher.finalize_reset(), Md5Stream::digest(message), "len {}", message.len());
+        }
+    }
+
+    /// Differential oracle against RustCrypto `md-5` called directly, i.e. not
+    /// through the backend module. With the default backend this is the same
+    /// implementation reached two ways; once another backend is compiled in
+    /// it becomes the independent reference that gates the switch.
+    mod differential {
+        use super::super::Md5Stream;
+        use proptest::prelude::*;
+
+        fn reference(data: &[u8]) -> [u8; 16] {
+            use md5::Digest as _;
+            md5::Md5::digest(data).into()
+        }
+
+        proptest! {
+            #![proptest_config(ProptestConfig::with_cases(256))]
+
+            #[test]
+            fn random_input_and_random_splits_match_reference(
+                data in proptest::collection::vec(any::<u8>(), 0..20_000),
+                splits in proptest::collection::vec(0usize..20_000, 0..8),
+            ) {
+                let want = reference(&data);
+                prop_assert_eq!(Md5Stream::digest(&data), want);
+
+                let mut cuts: Vec<usize> = splits.into_iter().map(|s| s % (data.len() + 1)).collect();
+                cuts.sort_unstable();
+                let mut hasher = Md5Stream::new();
+                let mut start = 0usize;
+                for cut in cuts {
+                    hasher.update(&data[start..cut]);
+                    start = cut;
+                }
+                hasher.update(&data[start..]);
+                prop_assert_eq!(hasher.finalize(), want);
+            }
+
+            #[test]
+            fn finalize_reset_matches_reference_for_each_message(
+                first in proptest::collection::vec(any::<u8>(), 0..3_000),
+                second in proptest::collection::vec(any::<u8>(), 0..3_000),
+            ) {
+                let mut hasher = Md5Stream::new();
+                hasher.update(&first);
+                prop_assert_eq!(hasher.finalize_reset(), reference(&first));
+                hasher.update(&second);
+                prop_assert_eq!(hasher.finalize(), reference(&second));
+            }
+        }
+    }
+}

crates/utils/src/lib.rs

@@ -36,6 +36,12 @@ pub mod io;
 #[cfg(feature = "hash")]
 pub mod hash;
 
+#[cfg(feature = "hash")]
+pub mod hash_lanes;
+
+#[cfg(feature = "hash")]
+pub mod hash_stream;
+
 #[cfg(feature = "os")]
 pub mod os;
 
@@ -59,6 +65,9 @@ pub mod dirs;
 #[cfg(feature = "hash")]
 pub use hash::*;
 
+#[cfg(feature = "hash")]
+pub use hash_stream::Md5Stream;
+
 #[cfg(feature = "io")]
 pub use io::*;
 

deny.toml

@@ -43,6 +43,10 @@ allow-git = [
     # RustFS fork carrying presigned expiry and constant-time authentication fixes.
     # owner: rustfs-maintainers  review: 2026-10
     # "https://github.com/rustfs/s3s.git",
+    # reatang fork of houseme/md5-simd carrying incremental SIMD `update_many`
+    # (houseme/md5-simd#1). Pinned by rev; switch to crates.io once released.
+    # owner: reatang  review: 2026-11
+    "https://github.com/reatang/md5-simd",
 ]
 
 [bans]

docs/testing/ci-gates.md

@@ -38,7 +38,7 @@ The aggregate requires the validation lanes already selected by `ci.yml`; this c
 | PR, non-doc change | `Workspace Test and Lint` | `ci.yml` `test-and-lint` | Via aggregate | `cargo clippy --all-targets -- -D warnings`; `cargo nextest run --profile ci --all --exclude e2e_test`; `cargo test --all --doc`; `scripts/check_migration_gate_count.sh` |
 | PR, non-doc change | `Typos` | `ci.yml` `typos` | Via aggregate | `typos` |
 | PR, non-doc change | `ILM Integration (serial)` | `ci.yml` `test-ilm-integration-serial` | Via aggregate | exact command in the job |
-| PR, non-doc change | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate | `cargo nextest run` with the job's `--features` |
+| PR, non-doc change | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)`, `Test and Lint (hash-md5-fast-md5)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate | `cargo nextest run` with the job's `--features` on the job's package set (`hash-md5-fast-md5` covers `rustfs-utils` + `rustfs-rio`: the alternate MD5 backend and its consumers) |
 | PR, non-doc change | `Connect Short Credential Boundary` | `ci.yml` `connect-short-credential-boundary` | Via aggregate | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail |
 | PR, non-doc change | `Build RustFS Debug Binary` | `ci.yml` `build-rustfs-debug-binary` | Via aggregate; prerequisite for black-box jobs | `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks` (binary plus its `rustfs.e2e.json` sidecar) |
 | PR, non-doc change | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Via aggregate | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |

rustfs/Cargo.toml

@@ -74,6 +74,10 @@ connect-e2e-short-credentials = []
 # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root.
 offline-enrollment-e2e-root = []
 rio-v2 = ["rustfs-ecstore/rio-v2"]
+# Alternate MD5 backend for ETag / Content-MD5 (crates/utils/src/hash_stream.rs). Not in `default` or `full`.
+hash-md5-fast-md5 = ["rustfs-utils/hash-md5-fast-md5", "rustfs-rio/hash-md5-fast-md5"]
+# Multi-stream SIMD MD5 for ETags: enables `RUSTFS_MD5_LANE_SERVER=on` (crates/rio/src/md5_lanes.rs). Trial wiring.
+hash-md5-simd = ["rustfs-utils/hash-md5-simd", "rustfs-rio/hash-md5-simd"]
 pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope"]
 # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.
 dial9 = ["rustfs-obs/dial9"]
@@ -342,7 +346,6 @@ const-str = { workspace = true, features = ["std", "proc"] }
 datafusion = { workspace = true, default-features = false, features = ["parquet", "recursive_protection", "sql"] }
 hex-simd.workspace = true
 matchit = { workspace = true }
-md-5 = { workspace = true }
 mime_guess = { workspace = true }
 percent-encoding = { workspace = true }
 pin-project-lite.workspace = true
@@ -380,6 +383,8 @@ libsystemd.workspace = true
 pyroscope = { workspace = true, optional = true, features = ["backend-pprof-rs"] }
 
 [dev-dependencies]
+# Independent MD5 reference for tests; production code goes through rustfs-utils.
+md-5 = { workspace = true }
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
 serial_test = { workspace = true }
 tempfile = { workspace = true }

rustfs/src/app/object/extract.rs

@@ -21,6 +21,7 @@ use crate::app::storage_api::object_usecase::concurrency::SNOWBALL_MEMBER_COMMIT
 use crate::app::storage_api::object_usecase::concurrency::SNOWBALL_STAGING_BYTES_LIMIT;
 use crate::app::trailer_adapter::trailer_source;
 use futures::stream::FuturesUnordered;
+use rustfs_utils::hash_stream::Md5Stream;
 use std::collections::HashSet;
 
 // One logical member can be preceded by local PAX, GNU long-name, and GNU
@@ -54,7 +55,8 @@ pin_project! {
     struct ExtractArchiveEtagReader<R> {
         #[pin]
         inner: R,
-        md5: Md5,
+        // `Some` until the EOF probe consumes it; `finished` guards every later poll.
+        md5: Option<Md5Stream>,
         expected_length: u64,
         bytes_read: u64,
         pending_final_byte: Option<u8>,
@@ -98,7 +100,7 @@ impl<R> ExtractArchiveEtagReader<R> {
     fn new(inner: R, expected_length: u64, state: Arc<Mutex<ExtractArchiveUploadState>>) -> Self {
         Self {
             inner,
-            md5: Md5::new(),
+            md5: Some(Md5Stream::new()),
             expected_length,
             bytes_read: 0,
             pending_final_byte: None,
@@ -155,8 +157,8 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
                         if let Ok(mut state) = this.state.lock()
                             && !state.body_complete
                         {
-                            state.etag =
-                                Some(hex_simd::encode_to_string(this.md5.clone().finalize(), hex_simd::AsciiCase::Lower));
+                            let digest = this.md5.take().unwrap_or_default().finalize();
+                            state.etag = Some(hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower));
                             state.body_complete = true;
                         }
                         *this.validating_eof = false;
@@ -180,7 +182,9 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
                         return Poll::Ready(Err(extract_archive_incomplete_body(*this.expected_length - *this.bytes_read)));
                     }
                     Poll::Ready(Ok(())) => {
-                        this.md5.update(final_buf.filled());
+                        if let Some(md5) = this.md5.as_mut() {
+                            md5.update(final_buf.filled());
+                        }
                         *this.bytes_read = match this.bytes_read.checked_add(1) {
                             Some(bytes_read) => bytes_read,
                             None => return Poll::Ready(Err(std::io::Error::other("archive read length overflow"))),
@@ -203,7 +207,9 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
                         return Poll::Ready(Err(extract_archive_incomplete_body(*this.expected_length - *this.bytes_read)));
                     }
                     Poll::Ready(Ok(())) => {
-                        this.md5.update(limited_buf.filled());
+                        if let Some(md5) = this.md5.as_mut() {
+                            md5.update(limited_buf.filled());
+                        }
                         limited_buf.filled().len()
                     }
                 }
@@ -2732,7 +2738,9 @@ impl DefaultObjectUsecase {
 #[cfg(test)]
 mod tests {
     use super::*;
+    // Independent MD5 reference: expectations must not come from the implementation under test.
     use http::{HeaderMap, HeaderName, HeaderValue};
+    use md5::{Digest as _, Md5};
     use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled};
     use tokio::io::AsyncReadExt;
     use tokio_tar::{Builder, EntryType, Header};

rustfs/src/app/object/get.rs

@@ -4828,6 +4828,7 @@ pub(super) async fn odm_get_from_source<S: OdmGetSource>(
 #[cfg(test)]
 mod on_demand_migration_tests {
     use super::*;
+    // Independent MD5 reference: expectations must not come from the implementation under test.
     use crate::on_demand_migration::{
         BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OdmStateError, OnDemandMigrationConfig, PathStyle, PolicyConfig,
         Provider, SourceConfig, SourceCredentials, SourceErrorPolicy, TlsConfig,
@@ -4837,6 +4838,7 @@ mod on_demand_migration_tests {
     };
     use async_trait::async_trait;
     use aws_sdk_s3::primitives::ByteStream as AwsByteStream;
+    use md5::{Digest as _, Md5};
     use std::collections::VecDeque;
     use std::sync::atomic::AtomicBool;
     use std::time::SystemTime;

rustfs/src/app/object/internal_put.rs

@@ -747,12 +747,14 @@ impl DefaultObjectUsecase {
 #[cfg(test)]
 mod tests {
     use super::*;
+    // Independent MD5 reference: expectations must not come from the implementation under test.
     use crate::app::storage_api::s3::{
         BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration,
         ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, Tag, VersioningConfiguration,
     };
     use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
+    use md5::{Digest as _, Md5};
     use rustfs_utils::http::{
         MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, SUFFIX_ODM_PULLED_AT, SUFFIX_ODM_SOURCE, SUFFIX_ODM_SOURCE_ETAG,
         SUFFIX_ODM_SOURCE_LAST_MODIFIED, SUFFIX_ODM_SOURCE_VERSION_ID, contains_key_str, get_consistent_str, get_str,

rustfs/src/app/object/mod.rs

@@ -124,7 +124,6 @@ use crate::table_catalog;
 use bytes::{BufMut as _, Bytes, BytesMut};
 use futures::{Stream, StreamExt};
 use http::{HeaderMap, HeaderValue, StatusCode};
-use md5::{Digest as Md5Digest, Md5};
 use metrics::{counter, histogram};
 use pin_project_lite::pin_project;
 use rustfs_audit::ObjectVersion as AuditObjectVersion;

rustfs/src/app/object/on_demand_migration_put.rs

@@ -297,6 +297,7 @@ impl OdmWriteBack for OnDemandMigrationWriteBack {
 #[cfg(test)]
 mod tests {
     use super::*;
+    // Independent MD5 reference: expectations must not come from the implementation under test.
     use crate::app::storage_api::multipart_usecase::contract::multipart::MultipartOperations as _;
     use crate::app::storage_api::s3::{
         BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration,
@@ -308,6 +309,7 @@ mod tests {
     use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
     use crate::on_demand_migration::{PullFailureReason, SourceSse};
     use http::Method;
+    use md5::Md5; // `sha2::Digest` below is the same trait and already in scope
     use rustfs_utils::http::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, contains_key_str, get_str};
     use sha2::{Digest as Sha256Digest, Sha256};
     use std::time::SystemTime;

rustfs/src/startup_runtime.rs

@@ -14,7 +14,9 @@
 
 use crate::{
     config::Config,
-    startup_runtime_hooks::{init_profiling_runtime, install_default_crypto_provider, log_startup_runtime_diagnostics},
+    startup_runtime_hooks::{
+        init_profiling_runtime, install_default_crypto_provider, install_md5_lane_server, log_startup_runtime_diagnostics,
+    },
     startup_tls_material::init_outbound_tls_material,
 };
 use rustfs_config::{DEFAULT_API_OBJECT_MAX_VERSIONS, ENV_API_OBJECT_MAX_VERSIONS};
@@ -26,6 +28,7 @@ pub(crate) async fn init_startup_runtime_foundation(config: &Config) -> Result<(
     init_profiling_runtime().await;
     rustfs_trusted_proxies::init();
     install_default_crypto_provider();
+    install_md5_lane_server();
     init_object_max_versions_config()?;
     init_outbound_tls_material(config).await
 }

rustfs/src/startup_runtime_hooks.rs

@@ -25,6 +25,7 @@ const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded";
 const LOG_SUBSYSTEM_LICENSE: &str = "license";
 const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_CRYPTO_PROVIDER_STATE: &str = "crypto_provider_state";
+const EVENT_MD5_LANE_SERVER_STATE: &str = "md5_lane_server_state";
 const EVENT_DIAL9_RUNTIME_STATUS: &str = "dial9_runtime_status";
 const EVENT_RUNTIME_LICENSE_STATUS: &str = "runtime_license_status";
 
@@ -119,6 +120,32 @@ pub(crate) fn install_default_crypto_provider() {
     }
 }
 
+/// ETag hashing: optionally move it from the request tasks to the multi-stream lane server.
+pub(crate) fn install_md5_lane_server() {
+    use rustfs_rio::md5_lanes::{LaneServerInstall, install_lane_server_from_env};
+    let (state, engine, threads, min_streams) = match install_lane_server_from_env() {
+        LaneServerInstall::Disabled => return,
+        LaneServerInstall::Unavailable => ("unavailable", "none", 0, 0),
+        LaneServerInstall::AlreadyInstalled => ("already_installed", "none", 0, 0),
+        LaneServerInstall::Installed {
+            engine,
+            threads,
+            min_streams,
+        } => ("installed", engine, threads, min_streams),
+    };
+    info!(
+        target: "rustfs::main",
+        event = EVENT_MD5_LANE_SERVER_STATE,
+        component = LOG_COMPONENT_MAIN,
+        subsystem = LOG_SUBSYSTEM_STARTUP,
+        state,
+        engine,
+        threads,
+        min_streams,
+        "MD5 lane server state checked"
+    );
+}
+
 pub(crate) async fn init_embedded_runtime_hooks(obs_endpoint: String) -> Result<()> {
     // Observability is a process-global tracing subscriber and can only be
     // installed once. A second embedded server in the same process reuses the

rustfs/src/storage/sse.rs

@@ -90,7 +90,6 @@ use chacha20poly1305::ChaCha20Poly1305;
 #[cfg(feature = "rio-v2")]
 use hmac::{Hmac, Mac};
 use http::{HeaderMap, HeaderName, HeaderValue};
-use md5::{Digest as Md5Digest, Md5};
 use rand::Rng;
 #[cfg(feature = "rio-v2")]
 use rand::RngExt;
@@ -148,9 +147,7 @@ const SEALED_KEY_SIZE: usize = DARE_HEADER_SIZE + 32 + DARE_TAG_SIZE;
 const OBJECT_KEY_DERIVATION_CONTEXT: &[u8] = b"object-encryption-key generation";
 
 fn md5_bytes(input: impl AsRef<[u8]>) -> [u8; 16] {
-    let mut hasher = Md5::new();
-    hasher.update(input.as_ref());
-    hasher.finalize().into()
+    rustfs_utils::hash_stream::Md5Stream::digest(input.as_ref())
 }
 
 fn md5_base64(input: impl AsRef<[u8]>) -> String {