zipatch-rs 1.5.0

//! Indexed apply: execute a pre-built [`Plan`] against a game install.
//!
//! [`IndexApplier`] is the apply-side counterpart of [`PlanBuilder`](crate::index::PlanBuilder).
//! It walks a [`Plan`]'s filesystem operations and per-target region timelines,
//! pulling source bytes from a caller-supplied [`PatchSource`] and writing
//! them to the install tree. The output is **byte-identical** to what the
//! sequential [`ZiPatchReader::apply_to`](crate::ZiPatchReader::apply_to)
//! driver produces for the same patch — see
//! `tests/index_apply_equivalence.rs` for the pin.
//!
//! # Reuse of [`ApplyContext`]
//!
//! `IndexApplier` is a thin shell over the existing [`ApplyContext`]: it
//! reuses the path cache, the 256-entry buffered file-handle cache, the
//! reusable [`flate2::Decompress`] state, and the observer. The only new
//! responsibility on this side is fetching source bytes by absolute patch
//! offset rather than streaming them out of an [`Iterator`] of parsed chunks.

use crate::Result;
use crate::ZiPatchError;
use crate::apply::observer::{ApplyObserver, ChunkEvent};
use crate::apply::path::{dat_path, generic_path, index_path};
use crate::apply::sqpk::{keep_in_remove_all, write_empty_block, write_zeros};
use crate::apply::{ApplyContext, ApplyMode};
use crate::index::plan::{
    FilesystemOp, PartSource, PatchSourceKind, Plan, Region, Target, TargetPath,
};
use crate::index::source::PatchSource;
use crate::index::verify::RepairManifest;
use crate::{Platform, apply::path::expansion_folder_id};
use flate2::{FlushDecompress, Status};
use std::fs;
use std::io::{Seek, SeekFrom, Write};
use std::ops::ControlFlow;
use std::path::PathBuf;
use tracing::{debug, debug_span, info, info_span, trace, warn};

/// Apply a [`Plan`] against a game install tree, fetching source bytes from
/// a [`PatchSource`].
///
/// Constructed via [`IndexApplier::new`]; configure platform/observer with
/// the `with_*` builder methods, then call [`IndexApplier::execute`] with a
/// `&Plan`. The plan's own [`Plan::platform`] is used by default; the
/// [`IndexApplier::with_platform`] override is only meaningful for synthetic
/// plans that don't carry a meaningful platform.
pub struct IndexApplier<S: PatchSource> {
    source: S,
    game_path: PathBuf,
    platform_override: Option<Platform>,
    mode: ApplyMode,
    observer: Option<Box<dyn ApplyObserver>>,
    checkpoint_sink: Option<Box<dyn crate::apply::CheckpointSink>>,
}

impl<S: PatchSource> IndexApplier<S> {
    /// Construct a new applier for `game_path`, pulling bytes from `source`.
    pub fn new(source: S, game_path: impl Into<PathBuf>) -> Self {
        Self {
            source,
            game_path: game_path.into(),
            platform_override: None,
            mode: ApplyMode::Write,
            observer: None,
            checkpoint_sink: None,
        }
    }

    /// Set the apply mode. See [`ApplyMode`].
    #[must_use]
    pub fn with_mode(mut self, mode: ApplyMode) -> Self {
        self.mode = mode;
        self
    }

    /// Override the platform pinned on the [`Plan`].
    ///
    /// Normally the plan's [`Plan::platform`] (taken from its
    /// `SqpkTargetInfo` chunk) is authoritative. Use this when applying a
    /// synthetic plan whose platform value is `Win32` by default but the
    /// target install actually expects PS3/PS4 path conventions.
    #[must_use]
    pub fn with_platform(mut self, platform: Platform) -> Self {
        self.platform_override = Some(platform);
        self
    }

    /// Install an [`ApplyObserver`] for progress reporting and cancellation.
    ///
    /// Per-target events are fired with kind `*b"IRGN"` (indexed-region
    /// boundary) — one event per [`Target`] rather than one per region, to
    /// avoid drowning the observer in events for plans with thousands of
    /// small regions. `bytes_read` is the cumulative count of bytes written
    /// across all targets and regions completed so far.
    ///
    /// # `'static` bound
    ///
    /// Mirrors [`ApplyContext::with_observer`](crate::ApplyContext::with_observer):
    /// the observer is boxed internally into a `Box<dyn ApplyObserver>`
    /// whose lifetime parameter defaults to `'static`. To pass an observer
    /// that holds a channel sender or similar handle, wrap it in
    /// `Arc<Mutex<...>>` or implement [`ApplyObserver`] on a struct that
    /// owns the handle directly.
    #[must_use]
    pub fn with_observer(mut self, observer: impl ApplyObserver + 'static) -> Self {
        self.observer = Some(Box::new(observer));
        self
    }

    /// Install a [`CheckpointSink`](crate::CheckpointSink) for apply-time
    /// checkpoint emission.
    ///
    /// Forwarded to the internal [`ApplyContext`] before [`Self::execute`]
    /// (or [`Self::execute_with_manifest`]) runs. The indexed driver emits
    /// one [`Checkpoint::Indexed`](crate::Checkpoint::Indexed) per target
    /// boundary and one every 64 regions inside a long target — the same
    /// cadence as the existing cancellation poll.
    ///
    /// Default is no sink: consumers that never call this method pay
    /// nothing.
    ///
    /// # Panics
    ///
    /// Panics if the sink reports
    /// [`CheckpointPolicy::FsyncEveryN`](crate::CheckpointPolicy::FsyncEveryN)
    /// with `n == 0`. Mirrors
    /// [`ApplyContext::with_checkpoint_sink`](crate::ApplyContext::with_checkpoint_sink)
    /// so both install paths surface the same diagnostic.
    #[must_use]
    pub fn with_checkpoint_sink(
        mut self,
        sink: impl crate::apply::CheckpointSink + 'static,
    ) -> Self {
        crate::apply::validate_checkpoint_policy(sink.policy());
        self.checkpoint_sink = Some(Box::new(sink));
        self
    }

    /// Execute the plan against the configured install.
    ///
    /// All [`FilesystemOp`]s in `plan.fs_ops` run first, in stream order,
    /// then per-target region writes. The handle cache is flushed before
    /// returning.
    ///
    /// # Errors
    ///
    /// Surfaces any [`ZiPatchError`] produced by the underlying I/O, the
    /// patch source, or the DEFLATE decompressor. Filesystem changes
    /// already applied are not rolled back.
    pub fn execute(self, plan: &Plan) -> Result<()> {
        self.resume_execute(plan, None).map(|_| ())
    }

    /// Resume a previously-interrupted indexed apply from an
    /// [`IndexedCheckpoint`](crate::IndexedCheckpoint).
    ///
    /// When `from` is `None`, behaves identically to [`Self::execute`]
    /// except for the return type: a successful run returns the final
    /// [`IndexedCheckpoint`](crate::IndexedCheckpoint) (with
    /// `next_target_idx` equal to the total number of targets in the plan
    /// and `next_region_idx == 0`).
    ///
    /// When `from` is `Some`, resume semantics:
    ///
    /// - Verify `from.plan_crc32` against [`Plan::crc32`]. Mismatch emits
    ///   a `warn!` and the resume falls through to a full apply from
    ///   `(target_idx=0, region_idx=0, fs_ops_done=false)` — same precedent
    ///   as the stale-manifest path in
    ///   [`Self::execute_with_manifest`] and the
    ///   `patch_name` / `patch_size` check in
    ///   [`crate::ZiPatchReader::resume_apply_to`].
    /// - If `from.fs_ops_done` is `true`, skip the `fs_ops` pass; otherwise
    ///   run every op in stream order (each op is idempotent w.r.t. the
    ///   install state the prior partial run would have left behind).
    /// - Fast-forward to `(from.next_target_idx, from.next_region_idx)`:
    ///   targets `0..next_target_idx` are skipped wholesale and inside
    ///   `next_target_idx`, regions `0..next_region_idx` are skipped. The
    ///   plan's region writes are offset-based and idempotent w.r.t. their
    ///   target offsets, so a re-run from `0` would produce the same
    ///   bytes; skipping is purely an optimisation, but a meaningful one
    ///   for plans with millions of regions.
    ///
    /// # Errors
    ///
    /// Same vocabulary as [`Self::execute`], plus
    /// [`ZiPatchError::SchemaVersionMismatch`] when `from.schema_version`
    /// does not equal [`crate::apply::IndexedCheckpoint::CURRENT_SCHEMA_VERSION`].
    /// The driver refuses to interpret a checkpoint whose layout this
    /// build cannot represent.
    #[allow(clippy::too_many_lines)]
    pub fn resume_execute(
        self,
        plan: &Plan,
        from: Option<&crate::apply::IndexedCheckpoint>,
    ) -> Result<crate::apply::IndexedCheckpoint> {
        if let Some(cp) = from {
            if cp.schema_version != crate::apply::IndexedCheckpoint::CURRENT_SCHEMA_VERSION {
                return Err(ZiPatchError::SchemaVersionMismatch {
                    kind: "indexed-checkpoint",
                    found: cp.schema_version,
                    expected: crate::apply::IndexedCheckpoint::CURRENT_SCHEMA_VERSION,
                });
            }
        }

        let plan_crc32 = plan.crc32();
        let region_count: usize = plan.targets.iter().map(|t| t.regions.len()).sum();
        let span = info_span!(
            "resume_execute",
            plan_crc32,
            targets = plan.targets.len(),
            regions = region_count,
            fs_ops = plan.fs_ops.len(),
        );
        let _enter = span.enter();
        let started = std::time::Instant::now();

        let effective_from = from.and_then(|cp| {
            if cp.plan_crc32 == plan_crc32 {
                Some(cp)
            } else {
                warn!(
                    expected_plan_crc32 = plan_crc32,
                    checkpoint_plan_crc32 = cp.plan_crc32,
                    "resume_execute: stale checkpoint, restarting from scratch"
                );
                None
            }
        });

        let skip_until_target = effective_from.map_or(0, |cp| cp.next_target_idx);
        let skip_until_region = effective_from.map_or(0, |cp| cp.next_region_idx);
        let bytes_written_in = effective_from.map_or(0, |cp| cp.bytes_written);
        let fs_ops_already_done = effective_from.is_some_and(|cp| cp.fs_ops_done);
        let resumed_from = effective_from.map(|cp| (cp.next_target_idx, cp.next_region_idx));

        let IndexApplier {
            mut source,
            game_path,
            platform_override,
            mode,
            observer,
            checkpoint_sink,
        } = self;

        let platform = platform_override.unwrap_or(plan.platform);
        let mut ctx = ApplyContext::new(game_path)
            .with_platform(platform)
            .with_mode(mode);
        if let Some(obs) = observer {
            ctx.observer = obs;
        }
        if let Some(sink) = checkpoint_sink {
            ctx.checkpoint_sink = sink;
        }

        if let Some((t, r)) = resumed_from {
            info!(
                plan_crc32,
                skipped_targets = t,
                skipped_regions = r,
                fs_ops_skipped = fs_ops_already_done,
                "resume_execute: resuming indexed apply"
            );
        }

        let mut bytes_written: u64 = bytes_written_in;
        let result: Result<()> = (|| {
            if fs_ops_already_done {
                debug!("resume_execute: fast-forwarded past fs_ops");
            } else {
                apply_fs_ops(&mut ctx, &plan.fs_ops)?;
            }
            // fs_ops have completed (either fast-forwarded or just run);
            // every checkpoint `apply_targets` emits from here carries
            // `fs_ops_done = true`.
            bytes_written = apply_targets(
                &mut ctx,
                &mut source,
                &plan.targets,
                plan_crc32,
                true,
                skip_until_target,
                skip_until_region,
                bytes_written_in,
            )?;
            Ok(())
        })();

        // Mirror `apply_to`: flush the handle cache on the way out so a
        // successful return implies all writes reached the OS, and so partial
        // progress is observable on the way out of an error path.
        let final_result = match ctx.flush() {
            Err(e) if result.is_ok() => Err(ZiPatchError::Io(e)),
            _ => result,
        };
        match final_result {
            Ok(()) => {
                info!(
                    bytes_written,
                    targets = plan.targets.len(),
                    resumed_from = ?resumed_from,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "apply_plan: indexed apply complete"
                );
                Ok(crate::apply::IndexedCheckpoint::new(
                    plan_crc32,
                    true,
                    plan.targets.len() as u64,
                    0,
                    bytes_written,
                ))
            }
            Err(e) => Err(e),
        }
    }

    /// Apply only the regions flagged in `manifest`, leaving every other byte
    /// of the install untouched.
    ///
    /// `plan.fs_ops` are **not** executed — the install is presumed to be
    /// already in the post-fs_ops state (that is the contract of having a
    /// non-empty `RepairManifest` against an otherwise-correct install).
    /// Per-target parent-directory creation still runs so a fully-missing
    /// target file can be recreated under a freshly-deleted subtree.
    ///
    /// # Stale manifest entries
    ///
    /// Manifest entries that reference a `target_idx` outside `plan.targets`,
    /// or a `region_idx` outside the addressed target's `regions`, are
    /// **silently skipped** — they do not raise an error. This is intentional:
    /// a `RepairManifest` is a plain `(target_idx, [region_idx]) -> _` map and
    /// callers may legitimately persist a manifest produced against an
    /// earlier plan revision and replay it against a freshly built plan
    /// whose target/region ordering has shifted. Skipping the stale entries
    /// lets the still-valid portion of the manifest apply cleanly; if a
    /// stricter check is needed, validate the manifest's indices against the
    /// current plan before calling this method.
    ///
    /// # Errors
    ///
    /// Same vocabulary as [`execute`](Self::execute).
    pub fn execute_with_manifest(self, plan: &Plan, manifest: &RepairManifest) -> Result<()> {
        let plan_crc32 = plan.crc32();
        let total_regions = manifest.total_missing_regions();
        let span = info_span!(
            "apply_plan",
            mode = "manifest",
            targets = manifest.missing_regions.len(),
            regions = total_regions,
        );
        let _enter = span.enter();
        let started = std::time::Instant::now();

        let IndexApplier {
            mut source,
            game_path,
            platform_override,
            mode,
            observer,
            checkpoint_sink,
        } = self;

        let platform = platform_override.unwrap_or(plan.platform);
        let mut ctx = ApplyContext::new(game_path)
            .with_platform(platform)
            .with_mode(mode);
        if let Some(obs) = observer {
            ctx.observer = obs;
        }
        if let Some(sink) = checkpoint_sink {
            ctx.checkpoint_sink = sink;
        }

        let mut bytes_written: u64 = 0;
        let result: Result<()> = (|| {
            bytes_written = apply_manifest_regions(
                &mut ctx,
                &mut source,
                plan,
                manifest,
                plan_crc32,
                true,
                0,
                0,
                0,
            )?;
            Ok(())
        })();

        let final_result = match ctx.flush() {
            Err(e) if result.is_ok() => Err(ZiPatchError::Io(e)),
            _ => result,
        };
        if final_result.is_ok() {
            info!(
                bytes_written,
                targets = manifest.missing_regions.len(),
                regions = total_regions,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "apply_plan: manifest replay complete"
            );
        }
        final_result
    }
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn apply_manifest_regions<S: PatchSource>(
    ctx: &mut ApplyContext,
    source: &mut S,
    plan: &Plan,
    manifest: &RepairManifest,
    plan_crc32: u32,
    fs_ops_done: bool,
    skip_until_target: u64,
    skip_until_region: u64,
    bytes_written_in: u64,
) -> Result<u64> {
    let mut scratch: Vec<u8> = Vec::new();
    let mut decompress_scratch: Vec<u8> = Vec::new();
    let mut bytes_written: u64 = bytes_written_in;
    let mut stale_targets: usize = 0;
    let mut stale_regions: usize = 0;

    // `missing_regions` is a `BTreeMap`, so iteration is already in ascending
    // `target_idx` order — that's the deterministic event/progress order we need.
    for (&target_idx, region_idxs) in &manifest.missing_regions {
        if (target_idx as u64) < skip_until_target {
            continue;
        }
        if ctx.observer.should_cancel() {
            debug!("apply_plan: cancelled at target boundary (manifest)");
            return Err(ZiPatchError::Cancelled);
        }
        emit_indexed_checkpoint(
            ctx,
            plan_crc32,
            fs_ops_done,
            target_idx as u64,
            0,
            bytes_written,
        )?;
        let Some(target) = plan.targets.get(target_idx) else {
            // Stale manifest from a different plan revision; ignore unknown
            // idx (documented contract on `execute_with_manifest`).
            stale_targets += 1;
            warn!(
                target_idx,
                "apply_plan: manifest target_idx out of range; skipping (stale manifest)"
            );
            continue;
        };

        let span = debug_span!(
            "apply_target",
            target_idx,
            path = %target_path_display(&target.path),
            regions = region_idxs.len(),
        );
        let _t_enter = span.enter();

        let path = resolve_target_path(ctx, &target.path)?;
        if let Some(parent) = path.parent() {
            ctx.ensure_dir_all(parent)?;
        }

        // `region_idxs` are sorted ascending at construction
        // (see `Verifier::execute`); only clone-and-sort when a caller has
        // hand-built a manifest in arbitrary order.
        let sorted: std::borrow::Cow<'_, [usize]> = if region_idxs.is_sorted() {
            std::borrow::Cow::Borrowed(region_idxs)
        } else {
            let mut v = region_idxs.clone();
            v.sort_unstable();
            std::borrow::Cow::Owned(v)
        };

        let mut last_end: Option<u64> = None;
        let skip_in_this_target = if (target_idx as u64) == skip_until_target {
            skip_until_region as usize
        } else {
            0
        };
        for (i, region_idx) in sorted.iter().enumerate() {
            if i < skip_in_this_target {
                last_end = None;
                continue;
            }
            // Cheap cancellation poll: every 64 regions, keeping the hot path
            // free of an extra branch on the per-region loop body.
            if i % 64 == 0 {
                if ctx.observer.should_cancel() {
                    debug!(target_idx, "apply_plan: cancelled mid-target (manifest)");
                    return Err(ZiPatchError::Cancelled);
                }
                if i > 0 {
                    emit_indexed_checkpoint(
                        ctx,
                        plan_crc32,
                        fs_ops_done,
                        target_idx as u64,
                        i as u64,
                        bytes_written,
                    )?;
                }
            }
            let Some(region) = target.regions.get(*region_idx) else {
                // Stale region index; skip rather than abort (documented
                // contract on `execute_with_manifest`).
                stale_regions += 1;
                warn!(
                    target_idx,
                    region_idx = *region_idx,
                    "apply_plan: manifest region_idx out of range; skipping (stale manifest)"
                );
                last_end = None;
                continue;
            };
            last_end = apply_region(
                ctx,
                source,
                &path,
                region,
                &mut scratch,
                &mut decompress_scratch,
                last_end,
            )?;
            bytes_written += u64::from(region.length);
        }

        debug!(
            target_idx,
            regions = region_idxs.len(),
            bytes_written,
            "apply_target: regions applied"
        );

        let event = ChunkEvent {
            index: target_idx,
            kind: *b"IRGN",
            bytes_read: bytes_written,
        };
        if let ControlFlow::Break(()) = ctx.observer.on_chunk_applied(event) {
            return Err(ZiPatchError::Cancelled);
        }
    }
    if stale_targets > 0 || stale_regions > 0 {
        warn!(
            stale_targets,
            stale_regions, "apply_plan: manifest had stale entries"
        );
    }
    Ok(bytes_written)
}

fn apply_fs_ops(ctx: &mut ApplyContext, ops: &[FilesystemOp]) -> Result<()> {
    if ops.is_empty() {
        return Ok(());
    }
    debug!(count = ops.len(), "apply_plan: applying fs_ops");
    for op in ops {
        match op {
            FilesystemOp::EnsureDir(rel) => {
                let path = ctx.game_path().join(rel);
                trace!(path = %path.display(), "fs_op: ensure dir");
                ctx.ensure_dir_all(&path)?;
            }
            FilesystemOp::MakeDirTree(rel) => {
                let path = ctx.game_path().join(rel);
                trace!(path = %path.display(), "fs_op: make dir tree");
                ctx.ensure_dir_all(&path)?;
            }
            FilesystemOp::DeleteDir(rel) => {
                let path = ctx.game_path().join(rel);
                if matches!(ctx.mode(), ApplyMode::DryRun) {
                    trace!(path = %path.display(), "fs_op: delete dir: dry-run, suppressed");
                } else {
                    fs::remove_dir(&path)?;
                    ctx.invalidate_dirs_created();
                    trace!(path = %path.display(), "fs_op: delete dir");
                }
            }
            FilesystemOp::DeleteFile(rel) => {
                let path = ctx.game_path().join(rel);
                // Flush and drop the cached handle before the OS unlink.
                ctx.evict_cached(&path)?;
                if matches!(ctx.mode(), ApplyMode::DryRun) {
                    trace!(path = %path.display(), "fs_op: delete file: dry-run, suppressed");
                    continue;
                }
                match fs::remove_file(&path) {
                    Ok(()) => trace!(path = %path.display(), "fs_op: delete file"),
                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                        // `DeleteFile` is used both as a user-intended delete
                        // (from `SqpkFile::DeleteFile`) and as a pre-truncate
                        // hint emitted by the builder for every `AddFile` at
                        // file_offset 0 — the sequential apply truncates in
                        // place via `set_len(0)` on the same write handle, but
                        // the indexed plan does not have that handle context
                        // available when applying fs_ops. NotFound is benign
                        // in both cases: the region writes that follow create
                        // the file (truncate-hint case), or the file genuinely
                        // does not exist (user-delete case, equivalent to
                        // `ignore_missing`).
                        trace!(path = %path.display(), "fs_op: delete file missing, ignored");
                    }
                    Err(e) => return Err(e.into()),
                }
            }
            FilesystemOp::RemoveAllInExpansion(expansion_id) => {
                // Match the sequential apply: flush all cached handles before
                // bulk-deleting expansion-folder contents.
                ctx.clear_file_cache()?;
                let folder = expansion_folder_id(*expansion_id);
                debug!(folder = %folder, "fs_op: remove all in expansion");
                for top in &["sqpack", "movie"] {
                    let dir = ctx.game_path().join(top).join(&folder);
                    if !dir.exists() {
                        continue;
                    }
                    for entry in fs::read_dir(&dir)? {
                        let path = entry?.path();
                        if path.is_file()
                            && !keep_in_remove_all(&path)
                            && matches!(ctx.mode(), ApplyMode::Write)
                        {
                            fs::remove_file(&path)?;
                        }
                    }
                }
            }
        }
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn apply_targets<S: PatchSource>(
    ctx: &mut ApplyContext,
    source: &mut S,
    targets: &[Target],
    plan_crc32: u32,
    fs_ops_done: bool,
    skip_until_target: u64,
    skip_until_region: u64,
    bytes_written_in: u64,
) -> Result<u64> {
    // Scratch buffers grown to the largest region we encounter; avoids
    // per-region allocations across the whole targets loop.
    let mut scratch: Vec<u8> = Vec::new();
    let mut decompress_scratch: Vec<u8> = Vec::new();
    let mut bytes_written: u64 = bytes_written_in;

    for (idx, target) in targets.iter().enumerate() {
        if (idx as u64) < skip_until_target {
            continue;
        }
        if ctx.observer.should_cancel() {
            debug!("apply_plan: cancelled at target boundary");
            return Err(ZiPatchError::Cancelled);
        }
        emit_indexed_checkpoint(ctx, plan_crc32, fs_ops_done, idx as u64, 0, bytes_written)?;
        let span = debug_span!(
            "apply_target",
            target_idx = idx,
            path = %target_path_display(&target.path),
            regions = target.regions.len(),
        );
        let _t_enter = span.enter();

        let path = resolve_target_path(ctx, &target.path)?;
        // Ensure parent directories exist. The sequential apply gets the
        // SqPack `sqpack/<expansion>/` folders via the patch's own
        // `AddDirectory` / `MakeDirTree` chunks (which land in `fs_ops`
        // and therefore run above us). Synthetic plans built in tests may
        // skip those, and the Generic-target `AddFile` path always wants
        // `ensure_dir_all(parent)`. Doing it here unconditionally is
        // idempotent thanks to the `dirs_created` cache on `ApplyContext`.
        if let Some(parent) = path.parent() {
            ctx.ensure_dir_all(parent)?;
        }

        let target_start_bytes = bytes_written;
        let mut last_end: Option<u64> = None;
        let skip_in_this_target = if (idx as u64) == skip_until_target {
            skip_until_region as usize
        } else {
            0
        };
        for (i, region) in target.regions.iter().enumerate() {
            if i < skip_in_this_target {
                // Fast-forward over already-applied regions in the resumed
                // target. The writes are offset-based and idempotent, so the
                // bytes are already on disk from the prior partial run, and
                // their contribution to `bytes_written` is already folded
                // into the checkpoint's recorded `bytes_written` (which we
                // seeded `bytes_written_in` from at entry). Skip without
                // re-adding so the running count stays consistent.
                last_end = None;
                continue;
            }
            // Mirror `apply_manifest_regions`: poll every 64 regions so a
            // single huge target with millions of regions stays cancellable.
            if i % 64 == 0 {
                if ctx.observer.should_cancel() {
                    debug!(target_idx = idx, "apply_plan: cancelled mid-target");
                    return Err(ZiPatchError::Cancelled);
                }
                if i > 0 {
                    emit_indexed_checkpoint(
                        ctx,
                        plan_crc32,
                        fs_ops_done,
                        idx as u64,
                        i as u64,
                        bytes_written,
                    )?;
                }
            }
            last_end = apply_region(
                ctx,
                source,
                &path,
                region,
                &mut scratch,
                &mut decompress_scratch,
                last_end,
            )?;
            bytes_written += u64::from(region.length);
        }

        debug!(
            target_idx = idx,
            regions = target.regions.len(),
            bytes_written_target = bytes_written - target_start_bytes,
            "apply_target: regions applied"
        );

        let event = ChunkEvent {
            index: idx,
            kind: *b"IRGN",
            bytes_read: bytes_written,
        };
        if let ControlFlow::Break(()) = ctx.observer.on_chunk_applied(event) {
            return Err(ZiPatchError::Cancelled);
        }
    }
    Ok(bytes_written)
}

// Render a `TargetPath` to a brief identifier for span fields. Cheap to format;
// avoids storing the full resolved `PathBuf` on every span entry.
//
// One allocation per call. Safe to invoke at per-target span entry (bounded
// by target count); do NOT call inside the per-region inner loop — that would
// allocate per region and put a `format!` on the hot path.
fn target_path_display(tp: &TargetPath) -> String {
    match tp {
        TargetPath::SqpackDat {
            main_id,
            sub_id,
            file_id,
        } => format!("sqpack:dat({main_id:x}/{sub_id:x}/{file_id})"),
        TargetPath::SqpackIndex {
            main_id,
            sub_id,
            file_id,
        } => format!("sqpack:index({main_id:x}/{sub_id:x}/{file_id})"),
        TargetPath::Generic(p) => p.clone(),
    }
}

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

// `last_end` is the writer's logical position after the previous region wrote,
// or `None` if the position is unknown (first region of a target, or the
// previous region was an `EmptyBlock` which seeks internally and leaves the
// cursor at the header offset rather than the region end). When the incoming
// region's `target_offset` equals `last_end`, the explicit seek is skipped —
// `BufWriter` continues appending into its in-memory buffer. Returns the new
// position when it's deterministic, or `None` after an `EmptyBlock`.
fn apply_region<S: PatchSource>(
    ctx: &mut ApplyContext,
    source: &mut S,
    path: &std::path::Path,
    region: &Region,
    scratch: &mut Vec<u8>,
    decompress_scratch: &mut Vec<u8>,
    last_end: Option<u64>,
) -> Result<Option<u64>> {
    let seek_needed = last_end != Some(region.target_offset);
    match &region.source {
        PartSource::Patch {
            patch_idx,
            offset,
            kind,
            decoded_skip,
        } => match *kind {
            PatchSourceKind::Raw { len } => {
                let len_us = len as usize;
                ensure_scratch(scratch, len_us);
                source.read(*patch_idx, *offset, &mut scratch[..len_us])?;
                let writer = ctx.open_cached(path)?;
                if seek_needed {
                    writer.seek(SeekFrom::Start(region.target_offset))?;
                }
                // Raw never sets decoded_skip; the kind's `len` already
                // matches `region.length` after any truncation.
                writer.write_all(&scratch[..len_us])?;
            }
            PatchSourceKind::Deflated {
                compressed_len,
                decompressed_len,
            } => {
                let comp_us = compressed_len as usize;
                ensure_scratch(scratch, comp_us);
                source.read(*patch_idx, *offset, &mut scratch[..comp_us])?;
                // Insert/refresh the cached handle, then split-borrow the
                // decompressor and the writer so both can be held across the
                // decompress loop.
                ctx.open_cached(path)?;
                let decompressor = &mut ctx.decompressor;
                let writer = ctx
                    .file_cache
                    .get_mut(path)
                    .expect("open_cached above inserted this path");
                if seek_needed {
                    writer.seek(SeekFrom::Start(region.target_offset))?;
                }
                decompress_into_sliced(
                    decompressor,
                    &scratch[..comp_us],
                    u64::from(*decoded_skip),
                    u64::from(region.length),
                    decompressed_len,
                    decompress_scratch,
                    writer,
                )?;
            }
        },
        PartSource::Zeros => {
            let writer = ctx.open_cached(path)?;
            if seek_needed {
                writer.seek(SeekFrom::Start(region.target_offset))?;
            }
            write_zeros(writer, u64::from(region.length))?;
        }
        PartSource::EmptyBlock { units } => {
            let writer = ctx.open_cached(path)?;
            write_empty_block(writer, region.target_offset, *units)?;
            // `write_empty_block` seeks twice internally and finishes with the
            // cursor at `target_offset + 20` (after the 20-byte header), not at
            // the region end — the next region cannot safely skip its seek.
            return Ok(None);
        }
        PartSource::Unavailable => {
            return Err(ZiPatchError::IndexSourceUnavailable {
                target_offset: region.target_offset,
                length: region.length,
            });
        }
    }
    Ok(Some(region.target_offset + u64::from(region.length)))
}

fn emit_indexed_checkpoint(
    ctx: &mut ApplyContext,
    plan_crc32: u32,
    fs_ops_done: bool,
    next_target_idx: u64,
    next_region_idx: u64,
    bytes_written: u64,
) -> Result<()> {
    let checkpoint = crate::apply::Checkpoint::Indexed(crate::apply::IndexedCheckpoint {
        schema_version: crate::apply::IndexedCheckpoint::CURRENT_SCHEMA_VERSION,
        plan_crc32,
        fs_ops_done,
        next_target_idx,
        next_region_idx,
        bytes_written,
    });
    debug!(
        next_target_idx,
        next_region_idx, bytes_written, "apply_plan: checkpoint recorded"
    );
    ctx.record_checkpoint(&checkpoint)
}

fn ensure_scratch(scratch: &mut Vec<u8>, needed: usize) {
    if scratch.len() < needed {
        scratch.resize(needed, 0);
    }
}

// Fully decompress one DEFLATE block into `scratch[..expected_out]`, reusing a
// caller-owned `flate2::Decompress` (reset on entry) and a caller-owned scratch
// `Vec<u8>` (grown as needed). Returns once the stream reaches `StreamEnd`,
// along with the actual number of bytes the decoder produced.
//
// Mirrors the semantics of `SqpkCompressedBlock::decompress_into_with` (raw
// DEFLATE, hand-rolled feed loop) but produces the whole decoded payload in
// one slice — required by [`crate::index::plan::Plan::compute_crc32`] and the
// `apply_region` call path alike, both of which slice into the decompressed
// output via `decoded_skip + region.length`.
//
// On `StreamEnd` we return whatever the decoder produced — even if it falls
// short of `expected_out` — to match the sequential apply's leniency in
// `SqpkCompressedBlock::decompress_into_with`. Real SqPack patches always
// produce exactly `expected_out` bytes; if a future malformed input ever
// produces fewer, the sequential path would write a short file and the
// indexed path must do the same to keep the equivalence pin. The returned
// `produced_total` lets callers bound their slice reads — `scratch` is
// reused across regions and is not zeroed between calls, so bytes past
// `produced_total` are stale and must never be read.
//
// Overshoot is structurally impossible: `decompressor.decompress` is given the
// slice `scratch[produced_total..needed]` as its output window and `flate2`
// caps `total_out` at the slice's length. When the decoder fills the window,
// the *next* iteration passes a zero-length slice and `BufError` fires with
// no forward progress, surfacing as a typed error. The overshoot branch the
// pre-polish code carried was unreachable and has been removed.
pub(crate) fn decompress_full(
    decompressor: &mut flate2::Decompress,
    input: &[u8],
    expected_out: u32,
    scratch: &mut Vec<u8>,
) -> Result<usize> {
    decompressor.reset(false);
    let needed = expected_out as usize;
    ensure_scratch(scratch, needed);
    let mut remaining = input;
    let mut produced_total: usize = 0;
    loop {
        let before_in = decompressor.total_in();
        let before_out = decompressor.total_out();
        let status = decompressor
            .decompress(
                remaining,
                &mut scratch[produced_total..needed],
                FlushDecompress::Finish,
            )
            .map_err(|e| {
                ZiPatchError::Decompress(std::io::Error::new(std::io::ErrorKind::InvalidData, e))
            })?;
        let consumed = (decompressor.total_in() - before_in) as usize;
        let produced = (decompressor.total_out() - before_out) as usize;
        produced_total += produced;
        remaining = &remaining[consumed..];
        match status {
            Status::StreamEnd => return Ok(produced_total),
            Status::Ok | Status::BufError => {
                if consumed == 0 && produced == 0 {
                    return Err(ZiPatchError::Decompress(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "DEFLATE stream made no forward progress",
                    )));
                }
            }
        }
    }
}

// Apply-side wrapper around `decompress_full`: decompress the whole block via
// the shared helper, then write the `[skip..skip + take]` slice of the decoded
// output to `w`. The slice window is `(decoded_skip, region.length)` from the
// caller; for the common case `skip == 0 && take == expected_out` this writes
// the entire decompressed stream verbatim.
fn decompress_into_sliced(
    decompressor: &mut flate2::Decompress,
    input: &[u8],
    skip: u64,
    take: u64,
    expected_out: u32,
    scratch: &mut Vec<u8>,
    w: &mut impl Write,
) -> Result<()> {
    let produced = decompress_full(decompressor, input, expected_out, scratch)?;
    let needed = expected_out as usize;
    let skip_us = usize::try_from(skip).map_err(|_| {
        ZiPatchError::Decompress(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("decoded_skip {skip} exceeds addressable range"),
        ))
    })?;
    let take_us = usize::try_from(take).map_err(|_| {
        ZiPatchError::Decompress(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("region length {take} exceeds addressable range"),
        ))
    })?;
    let end = skip_us.checked_add(take_us).ok_or_else(|| {
        ZiPatchError::Decompress(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "decoded_skip + region.length overflows usize",
        ))
    })?;
    if end > needed {
        return Err(ZiPatchError::Decompress(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!(
                "deflated region slice [{skip_us}..{end}] exceeds decompressed length {needed}"
            ),
        )));
    }
    // `scratch` is reused across regions without zeroing — clamp the read
    // window to what the decoder actually produced so stale bytes past
    // `produced` are never written. The sequential decoder achieves the
    // same effect by only writing `out[..produced]` per iteration.
    let clamped_end = end.min(produced);
    let clamped_start = skip_us.min(clamped_end);
    w.write_all(&scratch[clamped_start..clamped_end])?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::index::plan::{PartExpected, Region, Target, TargetPath};
    use crate::index::source::MemoryPatchSource;
    use flate2::Compression;
    use flate2::write::DeflateEncoder;

    fn dat_target(regions: Vec<Region>) -> Target {
        Target {
            path: TargetPath::SqpackDat {
                main_id: 0,
                sub_id: 0,
                file_id: 0,
            },
            final_size: regions
                .last()
                .map_or(0, |r| r.target_offset + u64::from(r.length)),
            regions,
        }
    }

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

    #[test]
    fn raw_region_writes_bytes_at_target_offset() {
        let payload = b"hello-raw-bytes!";
        let mut buf = vec![0u8; 1024];
        buf[100..100 + payload.len()].copy_from_slice(payload);
        let src = MemoryPatchSource::new(buf);

        let regions = vec![Region {
            target_offset: 50,
            length: payload.len() as u32,
            source: PartSource::Patch {
                patch_idx: 0,
                offset: 100,
                kind: PatchSourceKind::Raw {
                    len: payload.len() as u32,
                },
                decoded_skip: 0,
            },
            expected: PartExpected::SizeOnly,
        }];
        let plan = plan_with(vec![dat_target(regions)], vec![]);

        let tmp = tempfile::tempdir().unwrap();
        IndexApplier::new(src, tmp.path())
            .execute(&plan)
            .expect("apply must succeed");

        let target = tmp
            .path()
            .join("sqpack")
            .join("ffxiv")
            .join("000000.win32.dat0");
        let content = std::fs::read(&target).unwrap();
        assert_eq!(&content[50..50 + payload.len()], payload);
        assert!(content[..50].iter().all(|&b| b == 0));
    }

    #[test]
    fn deflated_region_decompresses_and_writes() {
        let raw: &[u8] = b"the quick brown fox jumps over the lazy dog";
        let mut enc = DeflateEncoder::new(Vec::new(), Compression::default());
        enc.write_all(raw).unwrap();
        let compressed = enc.finish().unwrap();

        let mut src_buf = vec![0u8; 4096];
        src_buf[200..200 + compressed.len()].copy_from_slice(&compressed);

        let regions = vec![Region {
            target_offset: 0,
            length: raw.len() as u32,
            source: PartSource::Patch {
                patch_idx: 0,
                offset: 200,
                kind: PatchSourceKind::Deflated {
                    compressed_len: compressed.len() as u32,
                    decompressed_len: raw.len() as u32,
                },
                decoded_skip: 0,
            },
            expected: PartExpected::SizeOnly,
        }];
        let plan = plan_with(vec![dat_target(regions)], vec![]);

        let tmp = tempfile::tempdir().unwrap();
        IndexApplier::new(MemoryPatchSource::new(src_buf), tmp.path())
            .execute(&plan)
            .expect("apply must succeed");

        let target = tmp
            .path()
            .join("sqpack")
            .join("ffxiv")
            .join("000000.win32.dat0");
        let content = std::fs::read(&target).unwrap();
        assert_eq!(&content[..raw.len()], raw);
    }

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

        let tmp = tempfile::tempdir().unwrap();
        IndexApplier::new(MemoryPatchSource::new(Vec::new()), tmp.path())
            .execute(&plan)
            .unwrap();

        let target = tmp
            .path()
            .join("sqpack")
            .join("ffxiv")
            .join("000000.win32.dat0");
        let content = std::fs::read(&target).unwrap();
        assert_eq!(content.len(), 64);
        assert!(content.iter().all(|&b| b == 0));
    }

    #[test]
    fn empty_block_region_matches_write_empty_block_output() {
        // Single-unit EmptyBlock at offset 0: 128 bytes total, with the
        // canonical 20-byte LE header at the start.
        let regions = vec![Region {
            target_offset: 0,
            length: 128,
            source: PartSource::EmptyBlock { units: 1 },
            expected: PartExpected::EmptyBlock { units: 1 },
        }];
        let plan = plan_with(vec![dat_target(regions)], vec![]);

        let tmp = tempfile::tempdir().unwrap();
        IndexApplier::new(MemoryPatchSource::new(Vec::new()), tmp.path())
            .execute(&plan)
            .unwrap();

        let target = tmp
            .path()
            .join("sqpack")
            .join("ffxiv")
            .join("000000.win32.dat0");
        let content = std::fs::read(&target).unwrap();
        assert_eq!(content.len(), 128);
        // Compare against the sequential apply's write_empty_block output.
        let mut cur = std::io::Cursor::new(Vec::<u8>::new());
        write_empty_block(&mut cur, 0, 1).unwrap();
        assert_eq!(content, cur.into_inner());
    }

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

        let tmp = tempfile::tempdir().unwrap();
        let err = IndexApplier::new(MemoryPatchSource::new(Vec::new()), tmp.path())
            .execute(&plan)
            .expect_err("Unavailable must abort");
        match err {
            ZiPatchError::IndexSourceUnavailable {
                target_offset,
                length,
            } => {
                assert_eq!(target_offset, 0);
                assert_eq!(length, 16);
            }
            other => panic!("expected IndexSourceUnavailable, got {other:?}"),
        }
    }

    #[test]
    fn decompress_into_sliced_does_not_leak_stale_scratch_bytes() {
        // Regression: `scratch` is reused across regions and never zeroed, so a
        // declared `expected_out` larger than the DEFLATE block actually
        // produces must not cause stale bytes from a prior decompression to be
        // written out. This test pre-fills the scratch with a sentinel payload
        // (simulating a previous, larger region), then asks
        // `decompress_into_sliced` to decode a short stream while *lying* about
        // its decompressed length. The output must contain only the bytes the
        // decoder produced — no sentinel bytes.
        let stale: Vec<u8> = (0..128u8).collect();
        let short_payload: &[u8] = b"short";

        let mut enc = DeflateEncoder::new(Vec::new(), Compression::default());
        enc.write_all(short_payload).unwrap();
        let compressed = enc.finish().unwrap();

        let mut scratch = stale.clone();
        let mut decompressor = flate2::Decompress::new(false);

        // Declare 64 bytes expected, but the stream only produces 5.
        let declared_decompressed: u32 = 64;
        let mut out = Vec::new();
        decompress_into_sliced(
            &mut decompressor,
            &compressed,
            0,
            u64::from(declared_decompressed),
            declared_decompressed,
            &mut scratch,
            &mut out,
        )
        .expect("short stream must still succeed (lenient on undershoot)");

        assert_eq!(
            out, short_payload,
            "output must only contain bytes the decoder actually produced; \
             any extra bytes are stale leftovers from prior scratch contents"
        );
    }

    #[test]
    fn deflated_short_stream_after_prior_region_does_not_corrupt_output() {
        // End-to-end variant of `decompress_into_sliced_does_not_leak_stale_scratch_bytes`:
        // exercise two consecutive Deflated regions through `IndexApplier::execute`
        // so the decompression scratch is shared across regions. The first
        // region fully fills the scratch with one payload; the second region
        // declares a larger decompressed_len than its DEFLATE block produces,
        // which on the pre-fix code would write trailing bytes from region 1
        // into region 2's slot. Asserts byte-for-byte that the second region's
        // file bytes contain only its own decoded payload.
        // Region 1's payload uses a high-byte sentinel pattern so it cannot
        // appear by coincidence in region 2's small payload or in any zero
        // pre-fill from the filesystem. The fingerprint also distinguishes
        // leaked bytes from any other content in the resulting file.
        let first_payload: Vec<u8> = (0..96).map(|i| 0x80u8 | (i as u8)).collect();
        let second_payload: &[u8] = b"abcd";

        let compress = |raw: &[u8]| {
            let mut enc = DeflateEncoder::new(Vec::new(), Compression::default());
            enc.write_all(raw).unwrap();
            enc.finish().unwrap()
        };
        let first_compressed = compress(&first_payload);
        let second_compressed = compress(second_payload);

        let mut src_buf = Vec::new();
        let first_offset = src_buf.len() as u64;
        src_buf.extend_from_slice(&first_compressed);
        let second_offset = src_buf.len() as u64;
        src_buf.extend_from_slice(&second_compressed);

        // The second region lies: declares decompressed_len far larger than the
        // stream actually produces (4 bytes). On pre-fix code, the extra
        // declared bytes would be read from stale scratch contents left over
        // from region 1. The region's `length` matches the lie, so the apply
        // attempts to write `declared_second_len` bytes total.
        let declared_second_len: u32 = first_payload.len() as u32;

        let regions = vec![
            Region {
                target_offset: 0,
                length: first_payload.len() as u32,
                source: PartSource::Patch {
                    patch_idx: 0,
                    offset: first_offset,
                    kind: PatchSourceKind::Deflated {
                        compressed_len: first_compressed.len() as u32,
                        decompressed_len: first_payload.len() as u32,
                    },
                    decoded_skip: 0,
                },
                expected: PartExpected::SizeOnly,
            },
            Region {
                target_offset: u64::from(first_payload.len() as u32),
                length: declared_second_len,
                source: PartSource::Patch {
                    patch_idx: 0,
                    offset: second_offset,
                    kind: PatchSourceKind::Deflated {
                        compressed_len: second_compressed.len() as u32,
                        decompressed_len: declared_second_len,
                    },
                    decoded_skip: 0,
                },
                expected: PartExpected::SizeOnly,
            },
        ];
        let plan = plan_with(vec![dat_target(regions)], vec![]);

        let tmp = tempfile::tempdir().unwrap();
        IndexApplier::new(MemoryPatchSource::new(src_buf), tmp.path())
            .execute(&plan)
            .expect("apply must succeed");

        let target = tmp
            .path()
            .join("sqpack")
            .join("ffxiv")
            .join("000000.win32.dat0");
        let content = std::fs::read(&target).unwrap();

        // Region 1 wrote its full payload verbatim.
        assert_eq!(
            &content[..first_payload.len()],
            first_payload.as_slice(),
            "region 1 must round-trip unchanged"
        );
        // Region 2: only the 4 bytes the decoder actually produced should
        // appear at the start of region 2's slot.
        let r2_start = first_payload.len();
        assert_eq!(
            &content[r2_start..r2_start + second_payload.len()],
            second_payload,
            "region 2's actually-produced bytes must round-trip"
        );
        // Post-fix: the apply only writes the bytes the decoder produced, so
        // the file ends at `r2_start + 4`. Pre-fix: the apply writes
        // `declared_second_len` bytes (region 1's stale tail), so the file
        // is longer and the stale bytes match `first_payload[4..]` byte for
        // byte. Both file length and the stale-byte equality are sufficient
        // independent signals.
        assert_eq!(
            content.len(),
            r2_start + second_payload.len(),
            "file must contain only what the decoder produced"
        );
        // Check region 2's slot only: pre-fix leaks `first_payload[4..]` at
        // `content[r2_start + 4..]`; post-fix never writes those bytes so the
        // file ends before that window even exists.
        let leak_pos = r2_start + second_payload.len();
        let leaked_tail = &first_payload[second_payload.len()..];
        assert!(
            content.len() < leak_pos + leaked_tail.len()
                || &content[leak_pos..leak_pos + leaked_tail.len()] != leaked_tail,
            "indexed apply must not leak stale scratch bytes from a prior region"
        );
    }

    /// C10: `decompress_full` must surface "DEFLATE stream made no forward
    /// progress" as a typed `Decompress` error when the decoder reports neither
    /// consumed nor produced bytes. Use a truncated DEFLATE stream — the
    /// decoder parses the valid prefix, then needs more input but the slice
    /// has been exhausted, so `BufError` fires with `consumed = produced = 0`.
    #[test]
    fn decompress_full_truncated_stream_surfaces_no_progress() {
        // Build a valid DEFLATE stream first, then truncate it before the
        // `StreamEnd` marker. Use a payload large enough that the prefix bytes
        // alone don't decode into the entire output.
        let raw: Vec<u8> = (0..256u32).map(|i| (i & 0xFF) as u8).collect();
        let mut enc = DeflateEncoder::new(Vec::new(), Compression::default());
        enc.write_all(&raw).unwrap();
        let full = enc.finish().unwrap();
        assert!(full.len() > 4, "stream must have at least a few bytes");
        // Cut the stream short — keep just enough bytes for the decoder to
        // accept the header but not enough to complete decompression.
        let truncated = &full[..full.len() / 2];

        let mut decompressor = flate2::Decompress::new(false);
        let mut scratch: Vec<u8> = Vec::new();
        let err = decompress_full(&mut decompressor, truncated, raw.len() as u32, &mut scratch)
            .expect_err("truncated stream must surface an error");

        // Two possible surface forms for the same underlying condition:
        //   1. The decoder consumes the truncated bytes, then on the next
        //      iteration both consumed and produced are 0 — our "no forward
        //      progress" arm fires. (zlib-rs / zlib-ng paths)
        //   2. The decoder raises an explicit `DataError` on a malformed
        //      stream — flate2 wraps it into Decompress error. (miniz_oxide
        //      may surface this directly.)
        // Both manifest as `ZiPatchError::Decompress(_)`; only the inner
        // message differs. Pin the typed variant; assert on the underlying
        // io::Error message only if it's the "no forward progress" arm.
        match err {
            ZiPatchError::Decompress(inner) => {
                let msg = inner.to_string();
                assert!(
                    msg.contains("no forward progress") || !msg.is_empty(),
                    "expected a meaningful Decompress error message, got {msg:?}"
                );
            }
            other => panic!("expected ZiPatchError::Decompress, got {other:?}"),
        }
    }

    #[test]
    fn delete_file_fs_op_removes_existing_file() {
        let tmp = tempfile::tempdir().unwrap();
        let target_rel = "victim.bin";
        let target_abs = tmp.path().join(target_rel);
        std::fs::write(&target_abs, b"to be removed").unwrap();
        assert!(target_abs.is_file());

        let plan = plan_with(vec![], vec![FilesystemOp::DeleteFile(target_rel.into())]);
        IndexApplier::new(MemoryPatchSource::new(Vec::new()), tmp.path())
            .execute(&plan)
            .unwrap();

        assert!(
            !target_abs.exists(),
            "DeleteFile fs_op must remove the file"
        );
    }
}