feat(verify): GREEN — parallel chunk decompression in verify()

verify() is now &self and decompresses chunks in parallel BATCHES (rayon
into_par_iter over threads*4 chunks), feeding them to the MD5/SHA1 hashers IN
ORDER — hashing is serial, but zlib decompression (the CPU cost of a full-image
hash) now uses every core. Splits a cacheless decompress_chunk out of
read_chunk so streaming the whole image neither pollutes nor contends the LRU.
Peak memory bounded to one batch of decompressed chunks.

Correctness validated by the existing differential oracle: e2e_dftt 27/27,
incl. exfat1/nps verify() computed-MD5 == stored MD5 (parallel decompression
reproduces the exact serial byte stream). New RED test green. Adds rayon to ewf.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: h4x0r

Cargo.lock

@@ -19,9 +19,10 @@ keywords = ["forensics", "ewf", "e01", "encase", "disk-image"]
 # members write `dep.workspace = true`. The inter-crate `ewf` version lives here
 # too, so a bump touches one line instead of every member.
 [workspace.dependencies]
-ewf = { path = "ewf", version = "0.2.1" }
+ewf = { path = "ewf", version = "0.2.2" }
 flate2 = "1"
 lru = "0.12"
+rayon = "1"
 thiserror = "2"
 log = "0.4"
 glob = "0.3"

Cargo.toml

@@ -21,6 +21,7 @@ verify = ["dep:md-5", "dep:sha-1"]
 [dependencies]
 flate2 = { workspace = true }
 lru = { workspace = true }
+rayon = { workspace = true }
 thiserror = { workspace = true }
 log = { workspace = true }
 glob = { workspace = true }

ewf/Cargo.toml

@@ -825,33 +825,45 @@ impl EwfReader {
     /// }
     /// ```
     #[cfg(feature = "verify")]
-    pub fn verify(&mut self) -> Result<VerifyResult> {
+    pub fn verify(&self) -> Result<VerifyResult> {
         use md5::Digest;
-
-        let has_sha1 = self.stored_sha1.is_some();
+        use rayon::prelude::*;
 
         let mut md5_hasher = md5::Md5::new();
-        let mut sha1_hasher = if has_sha1 {
+        let mut sha1_hasher = if self.stored_sha1.is_some() {
             Some(sha1::Sha1::new())
         } else {
             None
         };
 
-        self.position = 0;
-        let mut buf = vec![0u8; self.chunk_size as usize];
-        let mut remaining = self.total_size;
-
-        while remaining > 0 {
-            let to_read = std::cmp::min(remaining, buf.len() as u64) as usize;
-            let n = io::Read::read(self, &mut buf[..to_read])?;
-            if n == 0 {
-                break;
-            }
-            md5_hasher.update(&buf[..n]);
-            if let Some(ref mut h) = sha1_hasher {
-                h.update(&buf[..n]);
+        // Hashing is serial (MD5/SHA1 chain their state), but zlib decompression
+        // — the CPU cost of a full-image hash — is not. Decompress chunks in
+        // parallel BATCHES, then feed them to the hashers IN ORDER. A batch of
+        // `threads * 4` chunks keeps every core busy while bounding peak memory
+        // to one batch of decompressed chunks. `decompress_chunk` is cacheless,
+        // so streaming the whole image neither pollutes nor contends the LRU.
+        let chunk_count = self.chunks.len();
+        let batch = rayon::current_num_threads().saturating_mul(4).max(1);
+        let mut hashed: u64 = 0;
+
+        for start in (0..chunk_count).step_by(batch) {
+            let end = (start + batch).min(chunk_count);
+            let pages: Vec<Vec<u8>> = (start..end)
+                .into_par_iter()
+                .map(|ci| self.decompress_chunk(ci))
+                .collect::<Result<Vec<_>>>()?;
+            for page in pages {
+                // Trim the final chunk to the image's true length: a chunk
+                // decompresses into a full chunk_size buffer, but the last one
+                // may back fewer logical bytes.
+                let remaining = self.total_size.saturating_sub(hashed);
+                let take = (page.len() as u64).min(remaining) as usize;
+                md5_hasher.update(&page[..take]);
+                if let Some(ref mut h) = sha1_hasher {
+                    h.update(&page[..take]);
+                }
+                hashed += take as u64;
             }
-            remaining -= n as u64;
         }
 
         let computed_md5: [u8; 16] = md5_hasher.finalize().into();
@@ -891,6 +903,23 @@ impl EwfReader {
             }
         }
 
+        let page = self.decompress_chunk(chunk_id)?;
+
+        let mut cache = self
+            .cache
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        cache.put(chunk_id, page.clone());
+        Ok(page)
+    }
+
+    /// Decompress a chunk by index WITHOUT touching the cache.
+    ///
+    /// The cacheless counterpart to [`read_chunk`]: used by the parallel
+    /// [`verify`](Self::verify), which streams every chunk exactly once and so
+    /// must neither pollute the LRU nor serialize on its lock. Positioned read
+    /// + decompress, all through `&self`.
+    fn decompress_chunk(&self, chunk_id: usize) -> Result<Vec<u8>> {
         let mut page = vec![0u8; self.chunk_size as usize];
         let chunk = self.chunks[chunk_id].clone();
         let file = &self.segments[chunk.segment_idx];
@@ -924,11 +953,6 @@ impl EwfReader {
             }
         }
 
-        let mut cache = self
-            .cache
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        cache.put(chunk_id, page.clone());
         Ok(page)
     }