zipatch_rs/index/

verify.rs

//! Read-only verification of an install tree against a [`Plan`].
//!
//! [`PlanVerifier`] walks each [`Target`] in a [`Plan`], resolves it to a path in
//! the install tree, and inspects the bytes against the per-region
//! expectations the plan carries. The result is a [`RepairManifest`] naming
//! the targets and regions that need rewriting. Pair the manifest with
//! [`IndexApplier::execute_with_manifest`](crate::index::IndexApplier::execute_with_manifest)
//! to rewrite only the flagged regions without touching the rest of the install.
//!
//! # Verification policy
//!
//! - [`PartSource::Patch`] regions are size-only by default: the verifier
//!   checks that `target_offset + length <= file size on disk`. Without CRC the
//!   content inside a `Patch` region is **not** inspected — a single-byte flip
//!   in the middle of a `Patch` region is invisible. Call
//!   [`Plan::with_crc32`] before verifying to opt into content-level checks
//!   via [`PartExpected::Crc32`].
//! - [`PartSource::Zeros`] regions are content-checked: the verifier reads the
//!   range and flags any non-zero byte.
//! - [`PartSource::EmptyBlock`] regions are content-checked against the
//!   canonical payload the apply layer's internal `write_empty_block` helper
//!   would emit for the same `units` value (a 20-byte `SqPack` empty-block
//!   header followed by `units * 128 - 20` zero bytes).
//! - [`PartSource::Unavailable`] regions are always flagged — the builder does
//!   not emit them from any in-tree chunk parser, so encountering one means
//!   the plan is hand-built (or deserialized) with regions whose source bytes
//!   are unreachable; the verifier cannot repair them.

use crate::IndexResult as Result;
use crate::Platform;
use crate::apply::{ApplyConfig, ApplySession};
// Reuse the apply-layer path resolvers and empty-block writer so the verifier
// stays byte-identical to the applier by construction.
use crate::apply::path::{dat_path, generic_path, index_path};
use crate::apply::sqpk::empty_block_header;
use crate::index::plan::{PartExpected, PartSource, Plan, Region, Target, TargetPath};
#[cfg(feature = "parallel-verify")]
use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIterator};
use std::cell::RefCell;
use std::collections::BTreeMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use tracing::{debug, debug_span, info, info_span, trace};

const READ_BUF_CAPACITY: usize = 64 * 1024;

thread_local! {
    static REGION_SCRATCH: RefCell<Vec<u8>> = const { RefCell::new(Vec::new()) };
}

/// Per-target buckets of flagged regions and target-level diagnostics.
///
/// Produced by [`PlanVerifier::execute`]. The `missing_regions` map keys are
/// indices into [`Plan::targets`]; each value is a sorted ascending list of
/// indices into the target's [`Target::regions`].
///
/// # Stable iteration order
///
/// `missing_regions` is a [`BTreeMap`], so iteration is in ascending
/// `target_idx` order. Under the `serde` feature this also pins the serialized
/// key order — two manifests with identical contents serialize to identical
/// bytes regardless of the order they were populated in, which is required for
/// persisted-manifest workflows.
///
/// # Stability
///
/// `#[non_exhaustive]`: new diagnostic buckets (CRC-mismatched regions
/// separated from missing ones, permission-denied targets) may be added in
/// future minor versions. Consumers should treat the struct as
/// forward-compatible by name.
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct RepairManifest {
    /// `target_idx` → indices into [`Target::regions`] that failed verification.
    /// A fully-missing target file has every region index listed here. Iteration
    /// is ordered by `target_idx`.
    pub missing_regions: BTreeMap<usize, Vec<usize>>,
    /// Indices of targets whose underlying file does not exist on disk.
    pub missing_targets: Vec<usize>,
    /// Indices of targets whose underlying file exists but is shorter than
    /// [`Target::final_size`].
    pub size_mismatched: Vec<usize>,
}

impl RepairManifest {
    /// `true` iff no targets and no regions need repair.
    #[must_use]
    pub fn is_clean(&self) -> bool {
        self.missing_regions.is_empty()
            && self.missing_targets.is_empty()
            && self.size_mismatched.is_empty()
    }

    /// Sum of `missing_regions[k].len()` across all targets.
    #[must_use]
    pub fn total_missing_regions(&self) -> usize {
        self.missing_regions.values().map(Vec::len).sum()
    }
}

/// Walk a [`Plan`] against an install tree and produce a [`RepairManifest`].
///
/// Construct via [`PlanVerifier::new`], optionally override the platform with
/// [`PlanVerifier::with_platform`], then call [`PlanVerifier::execute`] with a `&Plan`.
/// The plan's [`Plan::platform`] is used by default.
///
/// The verifier never writes — it opens files read-only, reads only the
/// ranges it needs to inspect, and returns a manifest describing what is out
/// of shape. Pair it with
/// [`IndexApplier::execute_with_manifest`](crate::index::IndexApplier::execute_with_manifest)
/// to fix only the flagged regions.
pub struct PlanVerifier {
    install_root: PathBuf,
    platform_override: Option<Platform>,
}

impl PlanVerifier {
    /// Construct a verifier rooted at `install_root`.
    pub fn new(install_root: impl Into<PathBuf>) -> Self {
        Self {
            install_root: install_root.into(),
            platform_override: None,
        }
    }

    /// Override the platform pinned on the [`Plan`].
    #[must_use]
    pub fn with_platform(mut self, platform: Platform) -> Self {
        self.platform_override = Some(platform);
        self
    }

    /// Verify `plan` against the install tree.
    ///
    /// Returns a [`RepairManifest`] describing every target/region that needs
    /// rewriting. An empty manifest (see [`RepairManifest::is_clean`]) means
    /// the install matches the plan within the v1 policy: see the
    /// [module docs][crate::index::verify] for the per-source check matrix.
    ///
    /// # Patch-source caveat
    ///
    /// Single-byte damage inside a [`PartSource::Patch`]-sourced region is
    /// **not** detected by default — populate the plan's regions with
    /// [`PartExpected::Crc32`] via [`Plan::with_crc32`] to opt into
    /// content-level checks.
    ///
    /// # Errors
    ///
    /// Surfaces any [`crate::IndexError::Io`] produced while opening or
    /// reading an install-tree file, plus
    /// [`crate::IndexError::UnsupportedPlatform`] if the plan pins
    /// [`Platform::Unknown`] and the install contains `SqPack` targets.
    pub fn execute(self, plan: &Plan) -> Result<RepairManifest> {
        let span = info_span!(
            crate::tracing_schema::span_names::VERIFY_PLAN,
            targets = plan.targets.len()
        );
        let _enter = span.enter();
        let started = std::time::Instant::now();

        let PlanVerifier {
            install_root,
            platform_override,
        } = self;

        let platform = platform_override.unwrap_or(plan.platform);
        // Reuse ApplyConfig purely for its path-resolution caches; no writes
        // happen here. The handle cache stays empty.
        let mut ctx = ApplyConfig::new(install_root)
            .with_platform(platform)
            .into_session();

        let mut resolved: Vec<PathBuf> = Vec::with_capacity(plan.targets.len());
        for target in &plan.targets {
            resolved.push(resolve_target_path(&mut ctx, &target.path)?);
        }

        let parent = &span;
        // The chain body is identical for both branches; only the iterator
        // shape differs. With `parallel-verify` the per-target CRC/zero/empty
        // checks fan out across rayon's pool; without it everything runs
        // serially in the calling thread.
        #[cfg(feature = "parallel-verify")]
        let pair_iter = plan.targets.par_iter().zip(resolved.par_iter());
        #[cfg(not(feature = "parallel-verify"))]
        let pair_iter = plan.targets.iter().zip(resolved.iter());
        let outcomes: Vec<PerTargetOutcome> = pair_iter
            .enumerate()
            .map(|(idx, (target, path))| {
                parent.in_scope(|| {
                    let sub = debug_span!(
                        crate::tracing_schema::span_names::VERIFY_TARGET,
                        target = idx
                    );
                    let _e = sub.enter();
                    REGION_SCRATCH
                        .with(|cell| verify_target(idx, path, target, &mut cell.borrow_mut()))
                })
            })
            .collect::<Result<Vec<_>>>()?;

        let mut manifest = RepairManifest::default();
        for (idx, outcome) in outcomes.into_iter().enumerate() {
            match outcome {
                PerTargetOutcome::Missing => {
                    manifest.missing_targets.push(idx);
                    let region_count = plan.targets[idx].regions.len();
                    if region_count != 0 {
                        manifest
                            .missing_regions
                            .insert(idx, (0..region_count).collect());
                    }
                }
                PerTargetOutcome::Present {
                    size_mismatch,
                    flagged,
                } => {
                    if size_mismatch {
                        manifest.size_mismatched.push(idx);
                    }
                    if !flagged.is_empty() {
                        manifest.missing_regions.insert(idx, flagged);
                    }
                }
            }
        }
        // Stable ordering for both buckets and per-target region lists.
        manifest.missing_targets.sort_unstable();
        manifest.size_mismatched.sort_unstable();
        for v in manifest.missing_regions.values_mut() {
            v.sort_unstable();
        }
        info!(
            targets = plan.targets.len(),
            missing_targets = manifest.missing_targets.len(),
            size_mismatched = manifest.size_mismatched.len(),
            damaged_targets = manifest.missing_regions.len(),
            damaged_regions = manifest.total_missing_regions(),
            elapsed_ms = started.elapsed().as_millis() as u64,
            "verify_plan: scan complete"
        );
        Ok(manifest)
    }
}

enum PerTargetOutcome {
    Missing,
    Present {
        size_mismatch: bool,
        flagged: Vec<usize>,
    },
}

fn verify_target(
    idx: usize,
    path: &Path,
    target: &Target,
    scratch: &mut Vec<u8>,
) -> Result<PerTargetOutcome> {
    trace!(target = idx, path = %path.display(), "verify target");

    let Some(actual_size) = stat_size(path)? else {
        debug!(target = idx, path = %path.display(), "verify: target file missing");
        return Ok(PerTargetOutcome::Missing);
    };

    let size_mismatch = actual_size < target.final_size;
    if size_mismatch {
        debug!(
            target = idx,
            actual_size,
            final_size = target.final_size,
            "verify: target size mismatch"
        );
    }

    // Raw File rather than BufReader: every region is read after a fresh seek,
    // which would invalidate any BufReader's internal buffer on every call.
    // Each `read_exact` here is one syscall (or one streamed chunk for Zeros);
    // a 64 KiB BufReader would add a dead allocation per target with zero
    // amortisation benefit.
    let mut file = File::open(path)?;
    let mut flagged: Vec<usize> = Vec::new();

    for (region_idx, region) in target.regions.iter().enumerate() {
        if region_fails(region, actual_size, &mut file, scratch)? {
            flagged.push(region_idx);
        }
    }

    Ok(PerTargetOutcome::Present {
        size_mismatch,
        flagged,
    })
}

fn stat_size(path: &std::path::Path) -> Result<Option<u64>> {
    match std::fs::metadata(path) {
        Ok(meta) => Ok(Some(meta.len())),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(e) => Err(e.into()),
    }
}

fn resolve_target_path(ctx: &mut ApplySession, tp: &TargetPath) -> Result<PathBuf> {
    match *tp {
        TargetPath::SqpackDat {
            main_id,
            sub_id,
            file_id,
        } => dat_path(ctx, main_id, sub_id, file_id).map_err(Into::into),
        TargetPath::SqpackIndex {
            main_id,
            sub_id,
            file_id,
        } => index_path(ctx, main_id, sub_id, file_id).map_err(Into::into),
        TargetPath::Generic(ref rel) => Ok(generic_path(ctx, rel)),
    }
}

fn region_fails(
    region: &Region,
    actual_size: u64,
    file: &mut File,
    scratch: &mut Vec<u8>,
) -> Result<bool> {
    let len_u64 = u64::from(region.length);
    let end = region.target_offset.saturating_add(len_u64);
    if end > actual_size {
        return Ok(true);
    }

    // `PartExpected::Crc32` takes precedence: it is the only check that can
    // detect single-byte damage inside a `Patch`-sourced region. For Zeros /
    // EmptyBlock regions the per-source fast paths (no hash) are kept because
    // they're cheaper than a CRC over the same bytes.
    if let PartExpected::Crc32(expected) = region.expected {
        return check_crc32(file, region.target_offset, region.length, scratch, expected);
    }

    match region.source {
        PartSource::Patch { .. } => Ok(false),
        PartSource::Zeros => check_zeros(file, region.target_offset, len_u64, scratch),
        PartSource::EmptyBlock { units } => {
            check_empty_block(file, region.target_offset, units, scratch)
        }
        PartSource::Unavailable => Ok(true),
    }
}

fn check_crc32(
    file: &mut File,
    offset: u64,
    length: u32,
    scratch: &mut Vec<u8>,
    expected: u32,
) -> Result<bool> {
    let needed = length as usize;
    if scratch.len() < needed {
        scratch.resize(needed, 0);
    }
    file.seek(SeekFrom::Start(offset))?;
    file.read_exact(&mut scratch[..needed])?;
    Ok(crc32fast::hash(&scratch[..needed]) != expected)
}

fn check_zeros(file: &mut File, offset: u64, len: u64, scratch: &mut Vec<u8>) -> Result<bool> {
    if len == 0 {
        return Ok(false);
    }
    // Stream-check in 64 KiB chunks so multi-MB Zero runs do not balloon RAM.
    // Reuses the per-target scratch buffer so we never allocate on the hot path.
    if scratch.len() < READ_BUF_CAPACITY {
        scratch.resize(READ_BUF_CAPACITY, 0);
    }
    file.seek(SeekFrom::Start(offset))?;
    let mut remaining = len;
    while remaining > 0 {
        let take = remaining.min(READ_BUF_CAPACITY as u64) as usize;
        file.read_exact(&mut scratch[..take])?;
        if scratch[..take].iter().any(|&b| b != 0) {
            return Ok(true);
        }
        remaining -= take as u64;
    }
    Ok(false)
}

fn check_empty_block(
    file: &mut File,
    offset: u64,
    units: u32,
    scratch: &mut Vec<u8>,
) -> Result<bool> {
    if units == 0 {
        return Err(crate::IndexError::InvalidField {
            context: "EmptyBlock units must be non-zero",
        });
    }
    // Stream-compare the on-disk region against the canonical
    // (20-byte header + zeros) payload in fixed-size chunks. Avoids
    // materializing the up-to-4 GiB canonical buffer that a pathological
    // `units` value (near MAX_UNITS_PER_REGION) would otherwise demand,
    // and removes the per-`units` byte cache that used to dominate the
    // verifier's memory footprint. See issue #32.
    if scratch.len() < READ_BUF_CAPACITY {
        scratch.resize(READ_BUF_CAPACITY, 0);
    }
    file.seek(SeekFrom::Start(offset))?;
    let total = u64::from(units) * 128;
    let header = empty_block_header(units);
    let mut emitted: u64 = 0;
    let mut first = true;
    while emitted < total {
        let chunk_len = (total - emitted).min(READ_BUF_CAPACITY as u64) as usize;
        file.read_exact(&mut scratch[..chunk_len])?;
        if first {
            // total >= 128 (units >= 1) and READ_BUF_CAPACITY >= 128, so the
            // 20-byte header lives entirely in this first chunk.
            if scratch[..20] != header {
                return Ok(true);
            }
            if scratch[20..chunk_len].iter().any(|&b| b != 0) {
                return Ok(true);
            }
            first = false;
        } else if scratch[..chunk_len].iter().any(|&b| b != 0) {
            return Ok(true);
        }
        emitted += chunk_len as u64;
    }
    Ok(false)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::index::PatchRef;
    use crate::index::plan::{PartExpected, Region, Target, TargetPath};

    fn dat_target(regions: Vec<Region>, final_size: u64) -> Target {
        Target {
            path: TargetPath::SqpackDat {
                main_id: 0,
                sub_id: 0,
                file_id: 0,
            },
            final_size,
            regions,
        }
    }

    fn plan_with(targets: Vec<Target>) -> Plan {
        Plan {
            schema_version: Plan::CURRENT_SCHEMA_VERSION,
            platform: Platform::Win32,
            patches: vec![PatchRef {
                name: "synthetic".into(),
                patch_type: None,
            }],
            targets,
            fs_ops: vec![],
        }
    }

    #[test]
    fn repair_manifest_is_clean_when_empty() {
        let m = RepairManifest::default();
        assert!(m.is_clean());
        assert_eq!(m.total_missing_regions(), 0);
    }

    #[test]
    fn total_missing_regions_sums_per_target_buckets() {
        let mut m = RepairManifest::default();
        m.missing_regions.insert(0, vec![1, 2, 3]);
        m.missing_regions.insert(1, vec![4, 5, 6]);
        assert!(!m.is_clean());
        assert_eq!(m.total_missing_regions(), 6);
    }

    #[test]
    fn verifier_against_missing_target_flags_entire_target() {
        let regions = vec![
            Region {
                target_offset: 0,
                length: 16,
                source: PartSource::Zeros,
                expected: PartExpected::Zeros,
            },
            Region {
                target_offset: 16,
                length: 16,
                source: PartSource::Zeros,
                expected: PartExpected::Zeros,
            },
        ];
        let plan = plan_with(vec![dat_target(regions, 32)]);

        let tmp = tempfile::tempdir().unwrap();
        let manifest = PlanVerifier::new(tmp.path()).execute(&plan).unwrap();

        assert!(manifest.missing_targets.contains(&0));
        let regions = manifest
            .missing_regions
            .get(&0)
            .expect("missing target must populate every region");
        assert_eq!(regions, &vec![0, 1]);
    }

    fn canonical_empty_block_bytes(units: u32) -> Vec<u8> {
        let mut buf = vec![0u8; (units as usize) * 128];
        buf[0..4].copy_from_slice(&128u32.to_le_bytes());
        buf[12..16].copy_from_slice(&units.wrapping_sub(1).to_le_bytes());
        buf
    }

    fn write_to_temp(bytes: &[u8]) -> std::fs::File {
        use std::io::{Seek, Write};
        let mut f = tempfile::tempfile().unwrap();
        f.write_all(bytes).unwrap();
        f.seek(SeekFrom::Start(0)).unwrap();
        f
    }

    #[test]
    fn check_empty_block_accepts_canonical_payload() {
        // Exercises a `units` value that spans multiple read chunks
        // (8192 * 128 = 1 MiB > 64 KiB READ_BUF_CAPACITY).
        for units in [1u32, 4, 1024, 8192] {
            let mut f = write_to_temp(&canonical_empty_block_bytes(units));
            let mut scratch = Vec::new();
            let fails = check_empty_block(&mut f, 0, units, &mut scratch).unwrap();
            assert!(!fails, "units={units}: canonical payload must verify clean");
        }
    }

    #[test]
    fn check_empty_block_flags_corrupted_header() {
        let units = 4u32;
        let mut buf = vec![0u8; (units as usize) * 128];
        // No header bytes at all — first 20 bytes are zero instead of the
        // canonical `[128, 0, 0, units-1, 0]`.
        let mut f = write_to_temp(&buf);
        let mut scratch = Vec::new();
        let fails = check_empty_block(&mut f, 0, units, &mut scratch).unwrap();
        assert!(fails, "missing header must be flagged");

        // Also: header partially present but the `units - 1` field is wrong.
        buf[0..4].copy_from_slice(&128u32.to_le_bytes());
        buf[12..16].copy_from_slice(&999u32.to_le_bytes());
        let mut f = write_to_temp(&buf);
        let fails = check_empty_block(&mut f, 0, units, &mut scratch).unwrap();
        assert!(fails, "wrong units-1 field must be flagged");
    }

    #[test]
    fn check_empty_block_flags_corruption_in_zero_region() {
        let units = 8u32; // 1024-byte region; corruption past byte 20.
        let mut buf = canonical_empty_block_bytes(units);
        buf[500] = 0xFF;
        let mut f = write_to_temp(&buf);
        let mut scratch = Vec::new();
        let fails = check_empty_block(&mut f, 0, units, &mut scratch).unwrap();
        assert!(fails, "non-zero byte in body must be flagged");
    }

    #[test]
    fn check_empty_block_rejects_zero_units() {
        let mut f = tempfile::tempfile().unwrap();
        let mut scratch = Vec::new();
        let err = check_empty_block(&mut f, 0, 0, &mut scratch).unwrap_err();
        assert!(
            matches!(err, crate::IndexError::InvalidField { context } if context.contains("non-zero")),
            "got {err:?}"
        );
    }

    // --- Parallel fan-out determinism ---

    fn generic_target(rel: impl Into<String>, final_size: u64) -> Target {
        let region = Region {
            target_offset: 0,
            length: final_size as u32,
            source: PartSource::Zeros,
            expected: PartExpected::Zeros,
        };
        Target {
            path: TargetPath::Generic(rel.into()),
            final_size,
            regions: vec![region],
        }
    }

    // 36 Generic targets: 12 missing, 12 size-mismatched (smaller than declared),
    // 12 clean. Asserts that missing_targets, size_mismatched, and missing_regions
    // keys are all sorted ascending, and that two runs on equivalent input produce
    // equal manifests.
    #[test]
    fn parallel_fan_out_manifest_is_deterministic_and_sorted() {
        const TOTAL: usize = 36;
        const STRIPE: usize = TOTAL / 3; // 12 each
        let dir = tempfile::tempdir().unwrap();
        let mut targets = Vec::with_capacity(TOTAL);
        for i in 0..TOTAL {
            let rel = format!("tgt_{i:03}");
            if i < STRIPE {
                // Missing: no file on disk; declared size non-zero.
                targets.push(generic_target(&rel, 16));
            } else if i < 2 * STRIPE {
                // Size-mismatched: file shorter than declared final_size.
                let p = dir.path().join(&rel);
                std::fs::write(p, [0u8; 8]).unwrap();
                // declared 1 MiB, but file is only 8 bytes — region extends past EOF.
                targets.push(generic_target(&rel, 1024 * 1024));
            } else {
                // Clean: file matches declared size with zero content.
                let p = dir.path().join(&rel);
                std::fs::write(p, [0u8; 16]).unwrap();
                targets.push(generic_target(&rel, 16));
            }
        }
        let plan = plan_with(targets);

        let run1 = PlanVerifier::new(dir.path()).execute(&plan).unwrap();

        assert_eq!(run1.missing_targets.len(), STRIPE);
        assert_eq!(run1.size_mismatched.len(), STRIPE);

        for w in run1.missing_targets.windows(2) {
            assert!(w[0] < w[1], "missing_targets not sorted: {w:?}");
        }
        for w in run1.size_mismatched.windows(2) {
            assert!(w[0] < w[1], "size_mismatched not sorted: {w:?}");
        }
        for (key, regions) in &run1.missing_regions {
            for w in regions.windows(2) {
                assert!(w[0] < w[1], "missing_regions[{key}] not sorted: {w:?}");
            }
        }

        let run2 = PlanVerifier::new(dir.path()).execute(&plan).unwrap();
        assert_eq!(
            run1, run2,
            "two equivalent runs produced different manifests"
        );
    }

    // Builds a plan where targets are declared in a deliberately non-ascending
    // order relative to their expected outcome category. Asserts the sort
    // invariants still hold, guarding against a future merge-loop refactor that
    // drops the sort_unstable passes.
    #[test]
    fn parallel_fan_out_shuffled_target_order_manifest_sorted() {
        // Interleave missing and present targets in a non-trivial order.
        // Pattern (per 4): missing, clean, missing, size-mismatched.
        const GROUPS: usize = 8; // 32 targets total
        let dir = tempfile::tempdir().unwrap();
        let mut targets = Vec::with_capacity(GROUPS * 4);
        for g in 0..GROUPS {
            let base = g * 4;
            // missing
            targets.push(generic_target(format!("s_{base:03}"), 16));
            // clean
            let rel_c = format!("s_{:03}", base + 1);
            std::fs::write(dir.path().join(&rel_c), [0u8; 16]).unwrap();
            targets.push(generic_target(rel_c, 16));
            // missing
            targets.push(generic_target(format!("s_{:03}", base + 2), 16));
            // size-mismatched
            let rel_sm = format!("s_{:03}", base + 3);
            std::fs::write(dir.path().join(&rel_sm), [0u8; 4]).unwrap();
            targets.push(generic_target(rel_sm, 1024));
        }
        let plan = plan_with(targets);
        let manifest = PlanVerifier::new(dir.path()).execute(&plan).unwrap();

        for w in manifest.missing_targets.windows(2) {
            assert!(w[0] < w[1], "missing_targets not sorted: {w:?}");
        }
        for w in manifest.size_mismatched.windows(2) {
            assert!(w[0] < w[1], "size_mismatched not sorted: {w:?}");
        }
        for (key, regions) in &manifest.missing_regions {
            for w in regions.windows(2) {
                assert!(w[0] < w[1], "missing_regions[{key}] not sorted: {w:?}");
            }
        }
        // Sanity: expected counts.
        assert_eq!(manifest.missing_targets.len(), GROUPS * 2);
        assert_eq!(manifest.size_mismatched.len(), GROUPS);
    }

    // When at least one target causes a non-NotFound IO error, execute() must
    // return Err rather than swallowing it or hanging. We do not assert which
    // target's error wins (non-deterministic under rayon), only that the call
    // terminates with Err.
    #[cfg(target_family = "unix")]
    #[test]
    fn parallel_fan_out_propagates_io_error_from_one_target() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();

        // One readable file (clean) and one unreadable file (will cause File::open to fail).
        let rel_ok = "ok_target";
        let rel_err = "err_target";
        std::fs::write(dir.path().join(rel_ok), [0u8; 16]).unwrap();
        std::fs::write(dir.path().join(rel_err), [0u8; 16]).unwrap();
        std::fs::set_permissions(
            dir.path().join(rel_err),
            std::fs::Permissions::from_mode(0o000),
        )
        .unwrap();

        // Skip when running as root — CAP_DAC_OVERRIDE bypasses mode bits.
        if std::fs::File::open(dir.path().join(rel_err)).is_ok() {
            std::fs::set_permissions(
                dir.path().join(rel_err),
                std::fs::Permissions::from_mode(0o644),
            )
            .unwrap();
            eprintln!("skipping: running with CAP_DAC_OVERRIDE");
            return;
        }

        let targets = vec![generic_target(rel_ok, 16), generic_target(rel_err, 16)];
        let plan = plan_with(targets);
        let result = PlanVerifier::new(dir.path()).execute(&plan);

        std::fs::set_permissions(
            dir.path().join(rel_err),
            std::fs::Permissions::from_mode(0o644),
        )
        .unwrap();

        assert!(
            result.is_err(),
            "expected Err from unreadable target, got Ok"
        );
    }
}