girt 0.18.0

//! Additive batch publication. Retirement remains a separate verified operation.
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::{fs, io};

use super::{BatchLimits, MultiPackIndex, RepackBatch, WriteLimits, WriteOptions};
use crate::retention::MaintenanceIsolation;
use crate::{
    ObjectId, Objects, PackObject, PackWriteError, PackWriteLimits, PackWritten, ReadLimits,
    Repository,
};

/// Bounds each stage of additive incremental repacking.
///
/// By default, repository admission uses [`crate::OpenLimits::default`] under caller isolation,
/// bounding metadata, configuration, shallow data and reference stacks before selection or
/// publication. Admission failures preserve the opener's typed error in [`BatchRepackCause::Open`].
#[derive(Clone, Copy, Debug, Default)]
#[non_exhaustive]
pub struct BatchRepackLimits {
    /// Bounds source MIDX parsing and both source and post-publication pack snapshots.
    pub batch: BatchLimits,
    /// Per-object reconstruction bounds, including protected-copy verification.
    pub read: ReadLimits,
    /// Aggregate payload, object count and pack/index output bounds.
    pub write: PackWriteLimits,
    /// Bounds encoding the replacement MIDX, including duplicate object occurrences.
    pub index: WriteLimits,
    /// Bounds repository refresh before observing the source batch.
    pub open: crate::OpenLimits,
}

impl BatchRepackLimits {
    /// Replaces repository admission bounds, applied under isolation before publication.
    #[must_use]
    pub fn with_open(mut self, open: crate::OpenLimits) -> Self {
        self.open = open;
        self
    }
}

/// Furthest confirmed artifact visibility reached by an additive batch attempt.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[non_exhaustive]
pub enum BatchPublication {
    /// No checksum-named publication is confirmed; inspect the report's uncertain path.
    #[default]
    None,
    /// A complete pack is visible, but this attempt has not confirmed its index.
    Pack,
    /// A complete indexed pair is visible, including after a later MIDX failure.
    Pair,
}

/// Known observations and additive effects; no old artifact is removed.
#[derive(Clone, Debug, Default)]
#[non_exhaustive]
pub struct BatchRepackReport {
    /// Batch selected under this operation's isolation guard.
    pub batch: RepackBatch,
    /// Confirmed pack/index publication stage.
    pub publication: BatchPublication,
    /// Generated artifact statistics, available after private encoding completes.
    pub written: Option<PackWritten>,
    /// Whether the replacement MIDX was published, including a later sync/release failure.
    pub index_published: bool,
    /// Selected identities excluded after a protected primary copy was decoded and verified.
    pub protected_objects: usize,
    /// Payload bytes decoded from selected or protected copies under the input budget.
    pub verified_bytes: u64,
    /// Payload bytes decoded and identity-checked from the newly visible pair.
    pub replacement_verified_bytes: u64,
    /// Artifact whose publication may have completed before a failure was reported.
    pub uncertain: Option<PathBuf>,
}

/// Why an additive batch stopped; inspect its report before retrying under fresh isolation.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum BatchRepackCause {
    /// Caller exclusion could not be acquired.
    #[error("incremental repack isolation unavailable")]
    Isolation(#[source] io::Error),
    /// Repository metadata could not be refreshed.
    #[error("cannot refresh incremental repack repository")]
    Open(#[source] Box<crate::repository::OpenError>),
    /// Bounded MIDX batch observation failed.
    #[error("cannot select incremental repack batch")]
    Plan(#[source] Box<super::BatchError>),
    /// Source objects or pinned artifacts failed verification or reconstruction bounds.
    #[error("cannot read incremental repack input")]
    Read(#[source] Box<crate::ObjectReadError>),
    /// A selected identity is no longer readable from the pinned store.
    #[error("incremental repack object missing: {0}")]
    Missing(ObjectId),
    /// Encoding or an explicit write budget failed.
    #[error(transparent)]
    Write(#[from] PackWriteError),
    /// MIDX publication failed; its own error records index visibility.
    #[error("incremental repack MIDX publication failed")]
    Index(#[source] Box<super::InstallError>),
    /// Temporary storage, synchronization or publication failed.
    #[error(transparent)]
    Io(#[from] io::Error),
    /// An existing checksum-named artifact differs from the generated bytes.
    #[error("conflicting incremental repack artifact: {0}")]
    Existing(PathBuf),
    /// Repository layout, MIDX or protection markers changed during the attempt.
    #[error("incremental repack inputs changed")]
    Changed,
    /// Caller cancellation was observed; complete published artifacts remain.
    #[error("incremental repack cancelled")]
    Cancelled,
}

/// Failure retaining confirmed publication and verification effects.
#[derive(Debug, thiserror::Error)]
#[error("incremental batch repack failed")]
#[non_exhaustive]
pub struct BatchRepackFailure {
    /// Cause that stopped the operation.
    #[source]
    pub cause: BatchRepackCause,
    /// Effects that are not rolled back; reopen storage under isolation before retrying.
    pub report: Box<BatchRepackReport>,
}

impl From<crate::pack::publish::Error> for BatchRepackCause {
    fn from(error: crate::pack::publish::Error) -> Self {
        match error {
            crate::pack::publish::Error::Cancelled => Self::Cancelled,
            crate::pack::publish::Error::Existing(path) => Self::Existing(path),
            crate::pack::publish::Error::Io(error) => Self::Io(error),
        }
    }
}

impl Repository {
    /// Publishes an ordinary pack for one freshly selected incremental MIDX batch.
    ///
    /// Caller isolation spans repository refresh, selection, decoding and publication. Selection
    /// and copying share one pinned store; the advisory report from `plan_incremental_repack` is
    /// never accepted as execution input. Every selected identity is read with identity checks.
    /// An identity also indexed in any protected primary `.keep`, `.promisor` or `.mtimes` pack
    /// is excluded only after that protected copy decodes correctly. This includes unlisted packs,
    /// so duplicate ordinary copies cannot fold promisor data into the new ordinary pack.
    ///
    /// Payloads are retained under the aggregate input budget and encoded as full objects, without
    /// raw delta reuse, bitmap creation, network access or deletion. Existing source packs, loose
    /// files and sidecars remain intact. The new pair is synchronized and published pack first,
    /// then the replacement MIDX prefers it. Fewer than two eligible packs or no unprotected
    /// selected objects returns without publication. Repeating selection after ownership moves
    /// normally finds no useful second batch; old zero-owner packs remain for separate expiry.
    ///
    /// Cancellation and limits can leave a complete unindexed pack or indexed pair. See
    /// [`BatchRepackReport`] for confirmed effects. A successful return is not retirement
    /// authorization. The caller must continue to exclude dependent readers and pack writers
    /// before any later destructive operation.
    ///
    /// # Errors
    ///
    /// Returns [`BatchRepackFailure`] for exclusion refusal, invalid/unsupported source metadata,
    /// corruption, limits, cancellation, changed inputs or publication failures. No old storage is
    /// removed on any path. Pack-opening bounds also apply after publication and must accommodate
    /// the additional pair; a failure at that point retains `Pair` in the report.
    ///
    /// ```
    /// use std::sync::atomic::AtomicBool;
    ///
    /// use girt::multi_pack_index::{BatchPublication, BatchRepackLimits};
    /// use girt::retention::MaintenanceIsolation;
    /// use girt::{InitKind, ObjectFormat, Repository};
    ///
    /// // This single-threaded example exclusively owns its disposable repository.
    /// struct Private;
    /// impl MaintenanceIsolation for Private {
    ///     type Guard = ();
    ///     fn acquire(&mut self, _: &Repository) -> std::io::Result<()> {
    ///         Ok(())
    ///     }
    /// }
    /// let root = tempfile::tempdir()?;
    /// let repo = Repository::init(ObjectFormat::Sha1, root.path().join("r"), InitKind::Bare)?;
    /// let failure = repo
    ///     .repack_incremental_batch(
    ///         &mut Private,
    ///         1024 * 1024,
    ///         BatchRepackLimits::default(),
    ///         &AtomicBool::new(false),
    ///     )
    ///     .unwrap_err(); // No MIDX exists in this empty repository.
    /// assert_eq!(failure.report.publication, BatchPublication::None);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn repack_incremental_batch<I: MaintenanceIsolation>(
        &self,
        isolation: &mut I,
        batch_bytes: u64,
        limits: BatchRepackLimits,
        cancel: &AtomicBool,
    ) -> Result<BatchRepackReport, BatchRepackFailure> {
        self.repack_incremental_with(isolation, batch_bytes, limits, cancel, (|| {}, || {}))
    }

    fn repack_incremental_with<I: MaintenanceIsolation>(
        &self,
        isolation: &mut I,
        batch_bytes: u64,
        limits: BatchRepackLimits,
        cancel: &AtomicBool,
        checkpoints: (impl FnOnce(), impl FnOnce()),
    ) -> Result<BatchRepackReport, BatchRepackFailure> {
        let mut report = BatchRepackReport::default();
        let result = (|| -> Result<(), BatchRepackCause> {
            check(cancel)?;
            let _guard = isolation
                .acquire(self)
                .map_err(BatchRepackCause::Isolation)?;
            let current = crate::OpenOptions::default()
                .with_limits(limits.open)
                .open(self.git_dir(), cancel)
                .map_err(|error| BatchRepackCause::Open(Box::new(error)))?;
            if current.object_dir() != self.object_dir()
                || current.object_format() != self.object_format()
            {
                return Err(BatchRepackCause::Changed);
            }
            let _operation = crate::locks::operation();
            let prepared = super::batch::prepare(&current, batch_bytes, limits.batch, cancel)
                .map_err(|error| BatchRepackCause::Plan(Box::new(error)))?;
            let super::batch::PreparedBatch {
                objects,
                index,
                report: batch,
            } = prepared;
            report.batch = batch;
            if report.batch.packs.is_empty() {
                return Ok(());
            }
            let protection = protection(&objects, cancel)?;
            let directory_entries = pack_entries(&current, limits, cancel, None)?;
            let decoded = decode(
                &current,
                &objects,
                &index,
                &protection,
                limits,
                cancel,
                &mut report,
            )?;
            if decoded.is_empty() {
                return Ok(());
            }
            let input: Vec<_> = decoded
                .iter()
                .map(|(id, object)| PackObject {
                    id: *id,
                    kind: object.kind(),
                    data: object.data(),
                })
                .collect();
            let directory = current.object_dir().join("pack");
            let mut pack = tempfile::NamedTempFile::new_in(&directory)?;
            let mut idx = tempfile::NamedTempFile::new_in(&directory)?;
            let result = crate::pack::write::write_controlled(
                current.object_format(),
                &input,
                &mut pack,
                &mut idx,
                limits.write,
                crate::PackCompression::Ordinary,
                &mut || {
                    if cancel.load(Ordering::Relaxed) {
                        Err(PackWriteError::Io(io::ErrorKind::Interrupted.into()))
                    } else {
                        Ok(())
                    }
                },
            );
            check(cancel)?;
            let written = result?;
            report.written = Some(written);
            drop(input);
            drop(decoded);
            pack.as_file().sync_all()?;
            idx.as_file().sync_all()?;
            let base = directory.join(format!("pack-{}", written.checksum));
            let (after_pack, after_pair) = checkpoints;
            report.uncertain = Some(base.with_extension("pack"));
            let mut progress = crate::pack::publish::Progress::default();
            let mut after_pack = Some(after_pack);
            let publication = crate::pack::publish::pair_checked(
                crate::pack::publish::Pair {
                    pack,
                    index: idx,
                    basename: &base,
                    marker: None,
                },
                cancel,
                &mut progress,
                |step| {
                    if step == crate::pack::publish::Step::Index {
                        report.uncertain = Some(base.with_extension("idx"));
                        after_pack.take().expect("one index publication")();
                    }
                    if matches!(
                        step,
                        crate::pack::publish::Step::Pack | crate::pack::publish::Step::Index
                    ) {
                        fresh(
                            &current,
                            &objects,
                            &index,
                            &protection,
                            limits,
                            cancel,
                            (&directory_entries, &base),
                        )?;
                    }
                    Ok::<(), BatchRepackCause>(())
                },
            );
            report.publication = match progress.visible.last() {
                Some(path) if path.extension().is_some_and(|ext| ext == "idx") => {
                    BatchPublication::Pair
                }
                Some(_) => BatchPublication::Pack,
                None => BatchPublication::None,
            };
            if progress.uncertain.is_some() {
                report.uncertain = progress.uncertain;
            }
            publication?;
            report.publication = BatchPublication::Pair;
            report.uncertain = None;
            after_pair();
            fresh(
                &current,
                &objects,
                &index,
                &protection,
                limits,
                cancel,
                (&directory_entries, &base),
            )?;
            let published = verify_pair(&current, &base, limits, cancel, &mut report)?;
            // The shared publisher synchronized both names before a MIDX can refer to them.
            // The outer guard remains live; this private adapter avoids reacquiring a
            // potentially non-reentrant caller lock in the existing MIDX installer.
            let mut held = HeldIsolation;
            match current.write_multi_pack_index_controlled(
                &mut held,
                WriteOptions::default().with_preferred_pack(written.checksum),
                limits.index,
                limits.batch.packs,
                cancel,
            ) {
                Ok(_) => report.index_published = true,
                Err(error) => {
                    report.index_published = error.published;
                    return Err(BatchRepackCause::Index(Box::new(error)));
                }
            }
            published
                .check_unchanged()
                .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
            Ok(())
        })();
        match result {
            Ok(()) => Ok(report),
            Err(cause) => Err(BatchRepackFailure {
                cause,
                report: Box::new(report),
            }),
        }
    }
}

// Constructed only inside the lexical lifetime of the caller's acquired isolation guard.
struct HeldIsolation;
impl MaintenanceIsolation for HeldIsolation {
    type Guard = ();
    fn acquire(&mut self, _: &Repository) -> io::Result<()> {
        Ok(())
    }
}

struct Protection {
    path: PathBuf,
    flags: [bool; 3],
}

fn flags(path: &Path) -> io::Result<[bool; 3]> {
    let present = |extension| match fs::symlink_metadata(path.with_extension(extension)) {
        Ok(_) => Ok(true),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(error),
    };
    Ok([present("keep")?, present("promisor")?, present("mtimes")?])
}

fn protection(objects: &Objects, cancel: &AtomicBool) -> Result<Vec<Protection>, BatchRepackCause> {
    objects
        .primary_packs_with_paths()
        .into_iter()
        .map(|(path, pack)| {
            check(cancel)?;
            let flags = flags(path)?;
            if flags.iter().any(|flag| *flag) {
                pack.verify_index()
                    .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
            }
            Ok(Protection {
                path: path.to_owned(),
                flags,
            })
        })
        .collect()
}

fn decode(
    repository: &Repository,
    objects: &Objects,
    bytes: &[u8],
    protection: &[Protection],
    limits: BatchRepackLimits,
    cancel: &AtomicBool,
    report: &mut BatchRepackReport,
) -> Result<Vec<(ObjectId, crate::Object)>, BatchRepackCause> {
    let index =
        MultiPackIndex::parse_with(bytes, repository.object_format(), limits.batch.index)
            .map_err(|error| BatchRepackCause::Plan(Box::new(super::BatchError::Index(error))))?;
    let selected_paths: BTreeSet<_> = report
        .batch
        .packs
        .iter()
        .map(|pack| pack.path.clone())
        .collect();
    // The prepared MIDX already verified every claim, including unselected owners. Resolve
    // path membership once per bounded pack name rather than allocating a path for every object.
    let directory = repository.object_dir().join("pack");
    let selected = index
        .pack_names()
        .iter()
        .map(|name| {
            check(cancel)?;
            let path = directory.join(name).with_extension("pack");
            Ok(selected_paths.contains(path.as_path()))
        })
        .collect::<Result<Vec<bool>, BatchRepackCause>>()?;
    let primary = objects.primary_packs_with_paths();
    let mut decoded = Vec::new();
    for (position, bytes) in index
        .ids
        .chunks_exact(repository.object_format().digest_len())
        .enumerate()
    {
        check(cancel)?;
        let location = index
            .location(position)
            .map_err(|error| BatchRepackCause::Plan(Box::new(super::BatchError::Index(error))))?;
        if !selected[location.pack as usize] {
            continue;
        }
        let id = ObjectId::from_bytes(repository.object_format(), bytes)
            .map_err(|_| BatchRepackCause::Changed)?;
        if decoded.len() + report.protected_objects >= limits.write.max_objects as usize {
            return Err(PackWriteError::Limit("object count").into());
        }
        let remaining = limits
            .write
            .max_input_bytes
            .saturating_sub(report.verified_bytes);
        let header = objects
            .header(id, limits.read)
            .map_err(|error| BatchRepackCause::Read(Box::new(error)))?
            .ok_or(BatchRepackCause::Missing(id))?;
        if header.size() > remaining || header.size() > limits.write.max_object_bytes {
            return Err(PackWriteError::Limit("input bytes").into());
        }
        let mut protected = false;
        for ((_, pack), observation) in primary.iter().zip(protection) {
            if observation.flags.iter().any(|flag| *flag) {
                check(cancel)?;
                if let Some(position) = pack
                    .find(id)
                    .map_err(|error| BatchRepackCause::Read(Box::new(error)))?
                {
                    let object = pack
                        .read(position, limits.read, cancel)
                        .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
                    charge(object.data().len(), limits.write, report)?;
                    report.protected_objects += 1;
                    protected = true;
                    break;
                }
            }
        }
        if protected {
            continue;
        }
        let object = objects
            .read_controlled(id, limits.read, cancel)
            .map_err(|error| BatchRepackCause::Read(Box::new(error)))?
            .ok_or(BatchRepackCause::Missing(id))?;
        charge(object.data().len(), limits.write, report)?;
        decoded.push((id, object));
    }
    Ok(decoded)
}

fn charge(
    bytes: usize,
    limits: PackWriteLimits,
    report: &mut BatchRepackReport,
) -> Result<(), BatchRepackCause> {
    report.verified_bytes = report
        .verified_bytes
        .checked_add(bytes as u64)
        .filter(|bytes| *bytes <= limits.max_input_bytes)
        .ok_or(PackWriteError::Limit("input bytes"))?;
    Ok(())
}

fn fresh(
    repository: &Repository,
    objects: &Objects,
    index: &[u8],
    protection: &[Protection],
    limits: BatchRepackLimits,
    cancel: &AtomicBool,
    directory_observation: (&BTreeSet<PathBuf>, &Path),
) -> Result<(), BatchRepackCause> {
    check(cancel)?;
    objects
        .check_packs_unchanged(cancel)
        .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
    let directory = repository.object_dir().join("pack");
    let chain =
        match fs::symlink_metadata(directory.join("multi-pack-index.d/multi-pack-index-chain")) {
            Ok(_) => true,
            Err(error) if error.kind() == io::ErrorKind::NotFound => false,
            Err(error) => return Err(error.into()),
        };
    let (expected, base) = directory_observation;
    let mut expected = expected.clone();
    expected.remove(&base.with_extension("pack"));
    expected.remove(&base.with_extension("idx"));
    if chain
        || pack_entries(repository, limits, cancel, Some(base))? != expected
        || super::verify::read(
            &directory.join("multi-pack-index"),
            limits.batch.index.max_bytes,
        )
        .map_err(|error| BatchRepackCause::Plan(Box::new(super::BatchError::Index(error))))?
        .as_deref()
            != Some(index)
    {
        return Err(BatchRepackCause::Changed);
    }
    for observation in protection {
        check(cancel)?;
        if flags(&observation.path)? != observation.flags {
            return Err(BatchRepackCause::Changed);
        }
    }
    Ok(())
}

fn check(cancel: &AtomicBool) -> Result<(), BatchRepackCause> {
    if cancel.load(Ordering::Relaxed) {
        Err(BatchRepackCause::Cancelled)
    } else {
        Ok(())
    }
}

fn pack_entries(
    repository: &Repository,
    limits: BatchRepackLimits,
    cancel: &AtomicBool,
    output: Option<&Path>,
) -> Result<BTreeSet<PathBuf>, BatchRepackCause> {
    let mut paths = BTreeSet::new();
    let mut remaining = limits.batch.packs.max_directory_entries;
    for entry in fs::read_dir(repository.object_dir().join("pack"))? {
        check(cancel)?;
        remaining = remaining
            .checked_sub(1)
            .ok_or(PackWriteError::Limit("pack directory entries"))?;
        let path = entry?.path();
        if output.is_some_and(|base| {
            path == base.with_extension("pack") || path == base.with_extension("idx")
        }) {
            continue;
        }
        if path.extension().is_some_and(|extension| {
            ["pack", "idx", "keep", "promisor", "mtimes"]
                .iter()
                .any(|name| extension == *name)
        }) {
            paths.insert(path);
        }
    }
    Ok(paths)
}

fn verify_pair(
    repository: &Repository,
    base: &Path,
    limits: BatchRepackLimits,
    cancel: &AtomicBool,
    report: &mut BatchRepackReport,
) -> Result<crate::pack::FilePack, BatchRepackCause> {
    let mut opening = limits.batch.packs;
    let pack = crate::pack::FilePack::open(
        repository.object_format(),
        &base.with_extension("idx"),
        &mut opening,
        cancel,
    )
    .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
    pack.verify(cancel)
        .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
    for position in 0..pack.object_count() {
        check(cancel)?;
        let object = pack
            .read(position, limits.read, cancel)
            .map_err(|error| BatchRepackCause::Read(Box::new(error)))?;
        report.replacement_verified_bytes = report
            .replacement_verified_bytes
            .checked_add(object.data().len() as u64)
            .filter(|bytes| *bytes <= limits.write.max_input_bytes)
            .ok_or(PackWriteError::Limit("replacement verification bytes"))?;
    }
    Ok(pack)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::AtomicUsize;

    use super::*;
    use crate::maintenance_git as oracle;

    struct Isolation {
        held: Arc<AtomicBool>,
        acquisitions: Arc<AtomicUsize>,
    }
    struct Guard(Arc<AtomicBool>);
    impl Drop for Guard {
        fn drop(&mut self) {
            self.0.store(false, Ordering::Relaxed);
        }
    }
    impl MaintenanceIsolation for Isolation {
        type Guard = Guard;
        fn acquire(&mut self, _: &Repository) -> io::Result<Guard> {
            assert!(!self.held.swap(true, Ordering::Relaxed));
            self.acquisitions.fetch_add(1, Ordering::Relaxed);
            Ok(Guard(self.held.clone()))
        }
    }

    fn fixture() -> (tempfile::TempDir, Repository, Vec<PathBuf>, Isolation) {
        let root = tempfile::tempdir().unwrap();
        let repo = Repository::init(
            crate::ObjectFormat::Sha1,
            root.path().join("r"),
            crate::InitKind::Bare,
        )
        .unwrap();
        let mut paths = Vec::new();
        for payload in [b"first blob".as_slice(), b"second blob".as_slice()] {
            let id = repo.loose_objects().write_blob(payload).unwrap();
            let name = oracle::git(
                repo.git_dir(),
                &["pack-objects", "objects/pack/pack"],
                format!("{id}\n").as_bytes(),
            );
            paths.push(repo.object_dir().join(format!(
                "pack/pack-{}.pack",
                std::str::from_utf8(&name).unwrap().trim()
            )));
        }
        oracle::git(repo.git_dir(), &["multi-pack-index", "write"], b"");
        (
            root,
            repo,
            paths,
            Isolation {
                held: Arc::new(AtomicBool::new(false)),
                acquisitions: Arc::new(AtomicUsize::new(0)),
            },
        )
    }

    #[rstest::rstest]
    #[case(true, BatchPublication::Pack)]
    #[case(false, BatchPublication::Pair)]
    fn cancellation_preserves_confirmed_publication_and_one_guard(
        #[case] at_pack: bool,
        #[case] stage: BatchPublication,
    ) {
        let (_root, repo, paths, mut isolation) = fixture();
        let held = isolation.held.clone();
        let cancel = AtomicBool::new(false);
        let failure = repo
            .repack_incremental_with(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &cancel,
                (
                    || {
                        assert!(held.load(Ordering::Relaxed));
                        if at_pack {
                            cancel.store(true, Ordering::Relaxed);
                        }
                    },
                    || {
                        assert!(held.load(Ordering::Relaxed));
                        if !at_pack {
                            cancel.store(true, Ordering::Relaxed);
                        }
                    },
                ),
            )
            .unwrap_err();
        assert!(matches!(failure.cause, BatchRepackCause::Cancelled));
        assert_eq!(failure.report.publication, stage);
        assert_eq!(failure.report.uncertain.is_some(), at_pack);
        assert!(!failure.report.index_published);
        assert_eq!(isolation.acquisitions.load(Ordering::Relaxed), 1);
        assert!(!isolation.held.load(Ordering::Relaxed));
        for path in paths {
            assert!(path.exists());
        }
        oracle::git(repo.git_dir(), &["multi-pack-index", "verify"], b"");
        oracle::git(repo.git_dir(), &["fsck", "--full"], b"");
    }

    #[rstest::rstest]
    #[case("keep")]
    #[case("promisor")]
    #[case("mtimes")]
    fn changed_protection_between_names_refuses_index_publication(#[case] extension: &str) {
        let (_root, repo, paths, mut isolation) = fixture();
        let failure = repo
            .repack_incremental_with(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &AtomicBool::new(false),
                (
                    || {
                        fs::write(paths[0].with_extension(extension), b"changed protection")
                            .unwrap()
                    },
                    || {},
                ),
            )
            .unwrap_err();
        assert!(matches!(failure.cause, BatchRepackCause::Changed));
        assert_eq!(failure.report.publication, BatchPublication::Pack);
        let written = failure.report.written.unwrap();
        assert!(
            !repo
                .object_dir()
                .join(format!("pack/pack-{}.idx", written.checksum))
                .exists()
        );
        for path in paths {
            assert!(path.exists());
        }
    }

    #[test]
    fn new_pack_name_between_publications_refuses_index_publication() {
        let (_root, repo, paths, mut isolation) = fixture();
        let failure = repo
            .repack_incremental_with(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &AtomicBool::new(false),
                (
                    || {
                        fs::write(
                            repo.object_dir().join("pack/new-install.idx"),
                            b"synthetic concurrent install",
                        )
                        .unwrap()
                    },
                    || {},
                ),
            )
            .unwrap_err();
        assert!(matches!(failure.cause, BatchRepackCause::Changed));
        assert_eq!(failure.report.publication, BatchPublication::Pack);
        assert!(!failure.report.index_published);
        for path in paths {
            assert!(path.exists());
        }
    }

    #[test]
    fn damaged_replacement_pair_refuses_midx_publication() {
        let (_root, repo, paths, mut isolation) = fixture();
        let failure = repo
            .repack_incremental_with(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &AtomicBool::new(false),
                (
                    || {},
                    || {
                        let new = fs::read_dir(repo.object_dir().join("pack"))
                            .unwrap()
                            .map(|entry| entry.unwrap().path())
                            .find(|path| {
                                path.extension()
                                    .is_some_and(|extension| extension == "pack")
                                    && !paths.contains(path)
                            })
                            .unwrap();
                        let mut bytes = fs::read(&new).unwrap();
                        bytes[14] ^= 0xff;
                        fs::write(new, bytes).unwrap();
                    },
                ),
            )
            .unwrap_err();
        assert!(matches!(failure.cause, BatchRepackCause::Read(_)));
        assert_eq!(failure.report.publication, BatchPublication::Pair);
        assert!(!failure.report.index_published);
        for path in paths {
            assert!(path.exists());
        }
    }

    #[test]
    fn success_retains_one_isolation_guard_through_midx_publication() {
        let (_root, repo, paths, mut isolation) = fixture();
        let report = repo
            .repack_incremental_batch(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &AtomicBool::new(false),
            )
            .unwrap();
        assert_eq!(report.publication, BatchPublication::Pair);
        assert!(report.index_published);
        assert!(report.replacement_verified_bytes > 0);
        assert_eq!(isolation.acquisitions.load(Ordering::Relaxed), 1);
        assert!(!isolation.held.load(Ordering::Relaxed));
        for path in paths {
            assert!(path.exists());
        }
    }
    #[test]
    fn held_midx_lock_preserves_pair_and_reports_failed_index_update() {
        let (_root, repo, paths, mut isolation) = fixture();
        let lock = repo.object_dir().join("pack/multi-pack-index.lock");
        fs::write(&lock, b"other writer").unwrap();
        let failure = repo
            .repack_incremental_batch(
                &mut isolation,
                0,
                BatchRepackLimits::default(),
                &AtomicBool::new(false),
            )
            .unwrap_err();
        assert!(matches!(failure.cause, BatchRepackCause::Index(_)));
        assert_eq!(failure.report.publication, BatchPublication::Pair);
        assert!(!failure.report.index_published);
        assert_eq!(fs::read(&lock).unwrap(), b"other writer");
        for path in paths {
            assert!(path.exists());
        }
        oracle::git(repo.git_dir(), &["multi-pack-index", "verify"], b"");
        oracle::git(repo.git_dir(), &["fsck", "--full"], b"");
    }
}