Merge 0d2a6daa47 into d2d8ea7ec7

src/borg/archiver/compact_cmd.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from pathlib import Path
 
-from borgstore.store import ObjectNotFound as StoreObjectNotFound
+from borgstore.store import ItemInfo, ObjectNotFound as StoreObjectNotFound
 
 from ._common import with_repository
 from ..archive import Archive
@@ -11,7 +11,7 @@ from ..helpers import get_cache_dir
 from ..helpers.argparsing import ArgumentParser
 from ..constants import *  # NOQA
 from ..hashindex import ChunkIndex
-from ..helpers import set_ec, EXIT_ERROR, format_file_size, bin_to_hex
+from ..helpers import set_ec, EXIT_ERROR, format_file_size, bin_to_hex, hex_to_bin
 from ..helpers import ProgressIndicatorPercent
 from ..manifest import Manifest
 from ..repository import Repository
@@ -27,6 +27,7 @@ class ArchiveGarbageCollector:
         assert isinstance(repository, Repository)
         self.manifest = manifest
         self.chunks = None  # a ChunkIndex, here used for: id -> (is_used, stored_size)
+        self.missing_chunks = set()  # chunk ids referenced by archives but missing from the index
         self.total_files = None  # overall number of source files written to all archives in this repo
         self.total_size = None  # overall size of source file content data written to all archives
         self.archives_count = None  # number of archives
@@ -37,9 +38,10 @@ class ArchiveGarbageCollector:
 
     @property
     def repository_size(self):
-        if self.chunks is None or not self.stats:
+        if not self.stats:
             return None
-        return sum(entry.obj_size for id, entry in self.chunks.iteritems())  # sum of stored sizes
+        # on-disk size: sum of the actual pack file sizes, including bytes no index entry covers.
+        return sum(ItemInfo(*info).size for info in self.repository.store_list("packs"))
 
     def garbage_collect(self):
         """Removes unused chunks from a repository."""
@@ -160,7 +162,9 @@ class ArchiveGarbageCollector:
             for id in sorted(self.missing_chunks):
                 logger.debug(f"Missing object {bin_to_hex(id)}")
             set_ec(EXIT_ERROR)
-        if not self.dry_run:  # nuking removes the soft-deleted archives from the archives directory; skip on a dry run
+        # a missing chunk means the repo is damaged: keep the soft-deleted archives so "borg undelete"
+        # stays possible until "borg check --repair" has run.
+        if not self.dry_run and not self.missing_chunks:  # skip nuking on a dry run or a damaged repo
             logger.info("Cleaning archives directory from soft-deleted archives...")
             archive_infos = self.manifest.archives.list(sort_by=["ts"], deleted=True)
             for archive_info in archive_infos:
@@ -213,26 +217,78 @@ class ArchiveGarbageCollector:
 
         Two passes bound the memory use: the first keeps only per-pack byte counts to pick the packs
         to change, the second collects object ids for just those packs, not the whole index.
+
+        A pack's size is taken from the store (the actual file size), not from summing index entries.
+        A pack can hold bytes no index entry covers (a chunk copy stored again elsewhere, or objects
+        from a backup that crashed before writing its index); counting those against the file size lets
+        such packs be rewritten or dropped rather than kept forever. See issue #9868.
+
+        When the repo is damaged (archives reference chunks the index cannot find), packs holding
+        unindexed bytes are left untouched: those bytes may be live data "borg check --repair" can
+        recover. Packs without unindexed bytes are still compacted, so delete/prune waste is reclaimed.
         """
-        # Pass 1: one index scan, keep only per-pack byte tallies (two ints per pack, no id lists).
-        pack_total, pack_unused = defaultdict(int), defaultdict(int)
+        # Pass 1: list the pack files and their sizes; these are the packs we consider.
+        pack_total = {}  # pack_id -> file size in the store
+        for info in self.repository.store_list("packs"):
+            info = ItemInfo(*info)
+            pack_total[hex_to_bin(info.name)] = info.size
+
+        # sum the used bytes per pack from the index; also sum every entry's bytes (used or not) to
+        # detect which packs hold unindexed bytes. collect index entries whose pack is not in the store.
+        pack_used = defaultdict(int)
+        pack_indexed = defaultdict(int)  # pack_id -> bytes of all its index entries, used or not
+        stale_ids = []  # index entries referencing a pack file that is not in the store
+        stale_used = 0  # how many of those were still flagged used (lost data)
         for id, entry in self.chunks.iteritems():
             pid = entry.pack_id
-            pack_total[pid] += entry.obj_size
-            if not (entry.flags & ChunkIndex.F_USED):
-                pack_unused[pid] += entry.obj_size
+            if pid not in pack_total:
+                stale_ids.append(id)
+                if entry.flags & ChunkIndex.F_USED:
+                    stale_used += 1
+            else:
+                pack_indexed[pid] += entry.obj_size
+                if entry.flags & ChunkIndex.F_USED:
+                    pack_used[pid] += entry.obj_size
 
-        # decide each pack's fate from the tallies
+        if stale_ids:
+            drop_verb = "Would drop" if self.dry_run else "Dropping"
+            logger.warning(f"{drop_verb} {len(stale_ids)} index entries that reference missing pack files.")
+            if stale_used:
+                logger.error(f"{stale_used} of them were still in use: repository data is missing!")
+                set_ec(EXIT_ERROR)
+
+        # decide each pack's fate: unused = its file size minus the bytes the index says are used.
         drop_packs, rewrite_packs = set(), set()
+        pack_unused = {}  # pack_id -> wasted bytes, for the packs we act on (drop or rewrite)
+        skipped_unindexed = 0  # packs left alone because they hold unindexed bytes on a damaged repo
         for pid, total in pack_total.items():
-            unused = pack_unused.get(pid, 0)  # .get, not [pid]: don't insert all-used packs into the dict
+            used = pack_used.get(pid, 0)
+            unused = total - used
+            if unused < 0:
+                # the index says more is used than the file holds.
+                logger.error(f'Pack {bin_to_hex(pid)}: index claims more data than the file holds, run "borg check".')
+                set_ec(EXIT_ERROR)
+                continue  # leave this pack untouched
+            if self.missing_chunks and total > pack_indexed[pid]:
+                # the repo is damaged and this pack holds bytes no index entry covers. those bytes may be
+                # live data "borg check --repair" can recover from the pack headers, so leave the pack alone.
+                skipped_unindexed += 1
+                continue
             if unused == 0:
-                continue  # all used -> leave alone
-            if unused == total:
-                drop_packs.add(pid)  # all unused -> drop the whole file
+                continue  # fully used -> leave alone
+            if used == 0:
+                drop_packs.add(pid)  # no used bytes -> drop the whole file
+                pack_unused[pid] = unused
             elif 100 * unused / total >= self.threshold:
                 rewrite_packs.add(pid)  # mixed and wasteful enough -> copy survivors forward
+                pack_unused[pid] = unused
             # else: mixed but below threshold -> leave alone
+        if skipped_unindexed:
+            logger.error(
+                f"Skipped {skipped_unindexed} packs holding unindexed data that "
+                '"borg check --repair" might still recover; run it, then compact again.'
+            )
+            set_ec(EXIT_ERROR)
         if self.dry_run:
             freed = sum(pack_unused[pid] for pid in drop_packs) + sum(pack_unused[pid] for pid in rewrite_packs)
             logger.info(
@@ -240,7 +296,7 @@ class ArchiveGarbageCollector:
                 f"by dropping {len(drop_packs)} packs and rewriting {len(rewrite_packs)} packs."
             )
             return  # dry run: report only, change nothing
-        if not drop_packs and not rewrite_packs:
+        if not drop_packs and not rewrite_packs and not stale_ids:
             logger.info("Deleting 0 unused objects...")
             return  # nothing to reclaim; do not touch the cached chunk indexes
 
@@ -250,7 +306,7 @@ class ArchiveGarbageCollector:
         # Pass 2: collect object ids only for the affected packs (a subset, not the whole index)
         keep = {pid: set() for pid in rewrite_packs}  # survivors to copy forward, per pack
         drop = {pid: set() for pid in rewrite_packs}  # unused objects in those same packs
-        forget = []  # ids living in fully-unused packs we delete outright
+        forget = list(stale_ids)  # ids to remove from the index: stale entries, then dropped-pack objects
         for id, entry in self.chunks.iteritems():
             pid = entry.pack_id
             if pid in rewrite_packs:
@@ -258,10 +314,11 @@ class ArchiveGarbageCollector:
             elif pid in drop_packs:
                 forget.append(id)
 
-        # count what we remove: every object of a dropped pack, plus the unused objects cut from
-        # rewritten packs. unused objects in below-threshold packs stay, so they don't count.
+        # deleted counts index objects removed; reclaimed counts bytes freed, which also includes
+        # unindexed spans that are not index objects.
         deleted = len(forget) + sum(len(ids) for ids in drop.values())
-        logger.info(f"Deleting {deleted} unused objects...")
+        reclaimed = sum(pack_unused[pid] for pid in drop_packs) + sum(pack_unused[pid] for pid in rewrite_packs)
+        logger.info(f"Deleting {deleted} unused objects, freeing {format_file_size(reclaimed, iec=self.iec)}...")
         pi = ProgressIndicatorPercent(
             total=len(drop_packs) + len(rewrite_packs),
             msg="Compacting packs %3.1f%%",
@@ -312,6 +369,8 @@ class CompactMixIn:
             - use of ``borg delete`` or ``borg prune``
             - interrupted backups (consider retrying the backup before running compact)
             - backups of source files that encountered an I/O error mid-transfer and were skipped
+            - duplicate chunks written by concurrent backups (the redundant copies are unused)
+            - packs left behind by a backup that crashed before recording its objects
             - corruption of the repository (e.g., the archives directory lost entries; see notes below)
 
             You usually do not want to run ``borg compact`` after every write operation, but
@@ -330,8 +389,8 @@ class CompactMixIn:
             seeing fatal errors when creating backups or when archives are missing in
             ``borg repo-list``).
 
-            With ``--stats``, borg additionally reports the sum of stored object sizes
-            before and after compaction.
+            With ``--stats``, borg additionally reports the on-disk size of the pack files
+            before and after compaction (the reported compression factor is based on that size).
             """
         )
         subparser = ArgumentParser(parents=[common_parser], description=self.do_compact.__doc__, epilog=compact_epilog)

src/borg/repository.py

@@ -883,9 +883,7 @@ class Repository:
         # keep every object the chunk index lists for this pack, except the one being deleted.
         keep_ids = {cid for cid, e in self.chunks.iteritems() if e.pack_id == pack_id}
         keep_ids.discard(id)
-        # complete=False: a pack may also hold superseded bytes (an older copy of a chunk later re-put
-        # elsewhere, no longer in the index), so keep_ids and drop_ids need not cover the whole pack.
-        self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id}, complete=False)
+        self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id})
         if update_index:
             # close() only persists new entries incrementally, so write the full index here to record
             # the removal for the next borg process.
@@ -893,16 +891,20 @@ class Repository:
 
             write_chunkindex_to_repo(self, self.chunks, incremental=False, force_write=True, delete_other=True)
 
-    def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, complete: bool = True, chunks=None):
+    def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None):
         """Rewrite pack <pack_id>, keeping <keep_ids> and dropping <drop_ids>, then delete the old pack.
 
         keep_ids: chunk ids in this pack to copy into the new pack.
         drop_ids: chunk ids in this pack to discard. Must not overlap keep_ids.
-        complete: if True, keep_ids and drop_ids must together be every object in the pack (asserted).
-            If False, they may name only some of the pack's objects.
         chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index
             updates to. Must be the index keep_ids and drop_ids were derived from. Default: self.chunks.
 
+        keep_ids and drop_ids need not name every object in the pack. A pack can hold bytes that no
+        index entry covers: an older copy of a chunk that was later stored again elsewhere, or objects
+        from a backup that crashed before writing its index. Such bytes appear as gaps between the
+        listed objects and are dropped. An overlap between listed objects, or an object claiming to end
+        past the pack file, means index corruption and raises IntegrityError.
+
         Kept objects are copied into a new pack via store.defrag and repointed in the chunk index;
         dropped objects' chunk index entries are removed.
 
@@ -919,7 +921,7 @@ class Repository:
 
         assert keep_ids & drop_ids == set(), "an id cannot appear in both keep_ids and drop_ids"
 
-        # collect every object's range, tagged with whether it is kept, ordered by offset.
+        # collect every listed object's range, tagged with whether it is kept, ordered by offset.
         located = []  # (obj_offset, obj_id, obj_size, keep)
         for obj_id in keep_ids | drop_ids:
             keep = obj_id in keep_ids
@@ -928,21 +930,29 @@ class Repository:
             located.append((entry.obj_offset, obj_id, entry.obj_size, keep))
         located.sort()
 
-        # walk objects in offset order, collecting the survivors.
+        # walk objects in offset order, collecting the survivors. covered is the end offset of the last
+        # object; a gap (offset > covered) is unindexed bytes we drop, an overlap (offset < covered) is
+        # index corruption.
         kept = []  # (obj_offset, obj_id, obj_size), offset-ordered
         covered = 0
         for offset, obj_id, size, keep in located:
-            if complete:  # keep+drop are the whole pack: they must tile it with no gap or overlap
-                assert offset == covered, f"gap or overlap in pack {bin_to_hex(pack_id)} at offset {covered}"
-            covered += size
+            if offset < covered:
+                raise IntegrityError(
+                    f"pack {bin_to_hex(pack_id)}: overlapping objects at offset {offset} (index corruption), "
+                    f'run "borg check"'
+                )
+            covered = offset + size
             if keep:
                 kept.append((offset, obj_id, size))
 
-        if complete:  # the listed objects must reach the pack's end, no trailing object unaccounted for
-            pack_size = self.store.info(pack_key).size
-            assert (
-                covered == pack_size
-            ), f"pack {bin_to_hex(pack_id)}: {pack_size - covered} trailing bytes unaccounted for"
+        # reject an object that ends past the pack file: store.defrag would short-read it into a
+        # truncated object in the new pack, then the intact source pack is deleted.
+        pack_size = self.store.info(pack_key).size
+        if covered > pack_size:
+            raise IntegrityError(
+                f"pack {bin_to_hex(pack_id)}: object extends past end of file at offset {covered} "
+                f'(index corruption), run "borg check"'
+            )
 
         for drop_id in drop_ids:  # remove dropped objects from the index
             del chunks[drop_id]

src/borg/testsuite/archiver/compact_cmd_test.py

@@ -8,7 +8,7 @@ from ...helpers import get_cache_dir, bin_to_hex
 from ...hashindex import ChunkIndex
 from ...repository import Repository
 from ...cache import files_cache_name, discover_files_cache_names, list_chunkindex_hashes
-from ...cache import delete_chunkindex_from_repo
+from ...cache import delete_chunkindex_from_repo, write_chunkindex_to_repo
 from ...manifest import Manifest
 from . import cmd, create_regular_file, create_src_archive, generate_archiver_tests, open_repository, RK_ENCRYPTION
 from . import changedir
@@ -192,6 +192,210 @@ def test_compact_packs_respects_threshold(tmp_path):
         assert bin_to_hex(frugal_pack) in pack_names
 
 
+def test_compact_superseded_duplicate(tmp_path):
+    # Issue #9868 repro. Concurrent backups can write the same chunk into two packs; when the index
+    # fragments merge, only one copy stays indexed and the other is left as unindexed bytes in a pack.
+    # Once such a mixed pack crosses the threshold, compact must rewrite it, dropping those bytes.
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4  # one flush() -> one pack
+        # pack A: three objects W, X, Y (X will be the one later superseded by a copy in pack B)
+        for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX"), (H(2), b"YYYY")]:
+            repository.put(cid, fchunk(data, chunk_id=cid))
+        repository.flush()
+        pack_a = repository.chunks[H(0)].pack_id
+
+        # pack B: a second copy of X only, in its own pack (as a concurrent writer would have produced).
+        repository.put(H(1), fchunk(b"XXXX", chunk_id=H(1)))
+        repository.flush()
+        pack_b = repository.chunks[H(1)].pack_id
+        assert pack_b != pack_a
+        # after the (simulated) fragment merge, the index points X at pack B; pack A's X bytes are now
+        # a superseded, unindexed span. put() already repointed the index to pack B, so nothing to do.
+
+        # mark usage: W and X used, Y unused. pack A is now mixed (W used, X superseded gap, Y unused).
+        used = {H(0), H(1)}
+        for i in range(3):
+            entry = repository.chunks[H(i)]
+            flags = ChunkIndex.F_USED if H(i) in used else ChunkIndex.F_NONE
+            repository.chunks[H(i)] = entry._replace(flags=flags)
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        # W still readable; X still readable from pack B; Y dropped; pack A rewritten to only W's bytes.
+        assert pdchunk(repository.get(H(0))) == b"WWWW"
+        assert pdchunk(repository.get(H(1))) == b"XXXX"
+        assert repository.get(H(2), raise_missing=False) is None
+        assert bin_to_hex(pack_a) not in [info.name for info in repository.store_list("packs")]
+
+
+def test_compact_drops_orphan_pack(tmp_path):
+    # A pack with no index entries at all (a backup that crashed after writing the pack but before
+    # writing its index) must still be found and dropped: compact enumerates the actual pack files,
+    # not just the index (#9868).
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4
+        repository.put(H(0), fchunk(b"KEEP", chunk_id=H(0)))
+        repository.flush()
+        live_pack = repository.chunks[H(0)].pack_id
+        repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED)
+
+        # write an extra pack directly to the store, with NO chunk-index entry for it.
+        orphan_key = "packs/" + "ab" * 32
+        repository.store_store(orphan_key, b"orphan pack bytes")
+        assert "ab" * 32 in [info.name for info in repository.store_list("packs")]
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        pack_names = [info.name for info in repository.store_list("packs")]
+        assert "ab" * 32 not in pack_names  # orphan pack dropped
+        assert bin_to_hex(live_pack) in pack_names  # the used pack kept
+        assert pdchunk(repository.get(H(0))) == b"KEEP"
+
+
+def test_compact_superseded_waste_triggers_rewrite(tmp_path):
+    # A pack whose indexed objects are all used but which also holds unindexed bytes over the
+    # threshold must be rewritten (#9868). Delete one object's index entry so its bytes become
+    # unindexed waste, then compact must copy the used survivors into a fresh pack.
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4
+        for cid, data in [(H(0), b"AAAA"), (H(1), b"BBBBBBBBBB"), (H(2), b"CCCC")]:
+            repository.put(cid, fchunk(data, chunk_id=cid))
+        repository.flush()
+        pack = repository.chunks[H(0)].pack_id
+        # every remaining indexed object is used ...
+        for i in (0, 2):
+            repository.chunks[H(i)] = repository.chunks[H(i)]._replace(flags=ChunkIndex.F_USED)
+        # ... but H(1)'s big object becomes an unindexed superseded span (waste well over threshold).
+        del repository.chunks[H(1)]
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        new_pack = repository.chunks[H(0)].pack_id
+        assert new_pack != pack  # the pack was rewritten despite all indexed objects being used
+        assert bin_to_hex(pack) not in [info.name for info in repository.store_list("packs")]
+        assert pdchunk(repository.get(H(0))) == b"AAAA"
+        assert pdchunk(repository.get(H(2))) == b"CCCC"
+
+
+def test_compact_partial_when_chunks_missing(tmp_path):
+    # On a damaged repo (archives reference missing chunks) compact still reclaims ordinary waste from
+    # packs that hold no unindexed bytes, but leaves packs that DO hold unindexed bytes alone, since
+    # those bytes may be live data "borg check --repair" can recover (#9868).
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4
+        # clean pack: one used, one unused object -> reclaimable waste, but every byte is indexed.
+        for cid, data in [(H(0), b"KEEP"), (H(1), b"DROPME")]:
+            repository.put(cid, fchunk(data, chunk_id=cid))
+        repository.flush()
+        clean_pack = repository.chunks[H(0)].pack_id
+        repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED)
+        repository.chunks[H(1)] = repository.chunks[H(1)]._replace(flags=ChunkIndex.F_NONE)  # unused waste
+
+        # damaged pack: one used object plus unindexed bytes (H(3)'s entry dropped from the index).
+        for cid, data in [(H(2), b"LIVE"), (H(3), b"UNINDEXED")]:
+            repository.put(cid, fchunk(data, chunk_id=cid))
+        repository.flush()
+        damaged_pack = repository.chunks[H(2)].pack_id
+        repository.chunks[H(2)] = repository.chunks[H(2)]._replace(flags=ChunkIndex.F_USED)
+        del repository.chunks[H(3)]  # its bytes remain in damaged_pack as unindexed data
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.missing_chunks = {H(9)}  # mark the repo damaged
+        gc.compact_packs()
+
+        pack_names = [info.name for info in repository.store_list("packs")]
+        # clean pack: waste but no unindexed bytes -> compacted, survivor still readable
+        assert bin_to_hex(clean_pack) not in pack_names
+        assert pdchunk(repository.get(H(0))) == b"KEEP"
+        # damaged pack: holds unindexed bytes -> left untouched for check --repair
+        assert bin_to_hex(damaged_pack) in pack_names
+        assert pdchunk(repository.get(H(2))) == b"LIVE"
+
+
+def test_compact_skips_nuke_when_chunks_missing(archivers, request):
+    # On a damaged repo, compact must not finalize soft-deleted archives: it skips nuking so their
+    # directory entries survive until "borg check --repair" has run (#9868).
+    archiver = request.getfixturevalue(archivers)
+
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    create_src_archive(archiver, "kept")
+    create_src_archive(archiver, "gone")
+    cmd(archiver, "delete", "gone")  # soft-delete: finalized only once compact nukes it
+
+    # drop one used chunk's index entry and persist the index, so analyze_archives() reports a missing chunk.
+    repository = open_repository(archiver)
+    with repository:
+        some_id = next(iter(repository.chunks.iteritems()))[0]
+        del repository.chunks[some_id]
+        write_chunkindex_to_repo(repository, repository.chunks, incremental=False, force_write=True, delete_other=True)
+
+    output = cmd(archiver, "compact", "-v", exit_code=EXIT_ERROR)
+    assert "missing objects" in output  # the repo is seen as damaged ...
+    assert "Cleaning archives directory" not in output  # ... so soft-deleted archives are not nuked
+
+
+def test_compact_drops_stale_index_entries(tmp_path):
+    # An index entry whose pack file is gone from the store is stale: compact drops it. If it was still
+    # flagged used, that is lost data, so compact reports it and sets an error exit code (#9850).
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4
+        repository.put(H(0), fchunk(b"GONE", chunk_id=H(0)))
+        repository.flush()
+        gone_pack = repository.chunks[H(0)].pack_id
+        repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED)
+        repository.store_delete("packs/" + bin_to_hex(gone_pack))  # delete the pack file the index still references
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        assert H(0) not in repository.chunks  # stale entry dropped from the index
+
+
+def test_compact_skips_oversized_index_entry(tmp_path):
+    # An index entry claiming more bytes than its pack file holds means index corruption: compact must
+    # leave the pack untouched rather than rewrite it from a bad size (#9868).
+    from ...archiver.compact_cmd import ArchiveGarbageCollector
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        repository._pack_writer.max_count = 4
+        repository.put(H(0), fchunk(b"DATA", chunk_id=H(0)))
+        repository.flush()
+        pack = repository.chunks[H(0)].pack_id
+        entry = repository.chunks[H(0)]
+        repository.chunks[H(0)] = entry._replace(flags=ChunkIndex.F_USED, obj_size=entry.obj_size + 10000)
+
+        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        assert bin_to_hex(pack) in [info.name for info in repository.store_list("packs")]  # left untouched
+        assert pdchunk(repository.get(H(0))) == b"DATA"
+
+
 def test_compact_gc_after_index_loss(archivers, request):
     """When no cached chunk index exists (e.g. after an interrupted compact, #9748), compact
     rebuilds the index from the packs. The rebuilt entries must start unused (init_flags=F_NONE),

src/borg/testsuite/repository_test.py

@@ -397,9 +397,51 @@ def test_compact_pack_keep_all_is_noop(repo_fixtures, request):
         assert pdchunk(repository.get(H(1))) == b"DATA1"
 
 
-def test_compact_pack_complete_detects_short_coverage(repo_fixtures, request):
-    # complete=True must catch a pack whose listed objects do not reach its end: shrink the last
-    # object's recorded obj_size so the summed coverage falls short of the actual pack file size.
+def test_compact_pack_tolerates_gap(repo_fixtures, request):
+    # A pack may hold bytes no index entry covers (a superseded copy of a chunk re-put elsewhere).
+    # Drop the middle object's index entry to simulate that, then compact keeping the outer two:
+    # the gap must be skipped and its bytes reclaimed (issue #9868).
+    chunk0 = fchunk(b"DATA0", chunk_id=H(0))
+    chunk1 = fchunk(b"DATA1", chunk_id=H(1))
+    chunk2 = fchunk(b"DATA2", chunk_id=H(2))
+    repository = get_repository_from_fixture(repo_fixtures, request)
+    build_one_pack(repository, [(H(0), chunk0), (H(1), chunk1), (H(2), chunk2)])
+    with repository:
+        old_pack_id = repository.chunks[H(0)].pack_id
+        del repository.chunks[H(1)]  # H(1)'s bytes stay in the pack but are now unindexed (a gap)
+
+        new_pack_id = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set())
+
+        assert new_pack_id is not None and new_pack_id != old_pack_id
+        assert pdchunk(repository.get(H(0))) == b"DATA0"
+        assert pdchunk(repository.get(H(2))) == b"DATA2"
+        packs = {info.name: info.size for info in repository.store_list("packs")}
+        assert bin_to_hex(old_pack_id) not in packs
+        assert packs[bin_to_hex(new_pack_id)] == len(chunk0) + len(chunk2)  # gap bytes reclaimed
+
+
+def test_compact_pack_tolerates_trailing_bytes(repo_fixtures, request):
+    # Same as the gap case, but the unindexed bytes are at the end of the pack: dropping the last
+    # object's index entry leaves trailing bytes the listed objects do not reach. Must still succeed.
+    chunk0 = fchunk(b"DATA0", chunk_id=H(0))
+    chunk1 = fchunk(b"DATA1", chunk_id=H(1))
+    repository = get_repository_from_fixture(repo_fixtures, request)
+    build_one_pack(repository, [(H(0), chunk0), (H(1), chunk1)])
+    with repository:
+        old_pack_id = repository.chunks[H(0)].pack_id
+        del repository.chunks[H(1)]  # trailing unindexed bytes
+
+        new_pack_id = repository.compact_pack(old_pack_id, keep_ids={H(0)}, drop_ids=set())
+
+        assert new_pack_id is not None and new_pack_id != old_pack_id
+        assert pdchunk(repository.get(H(0))) == b"DATA0"
+        packs = {info.name: info.size for info in repository.store_list("packs")}
+        assert packs[bin_to_hex(new_pack_id)] == len(chunk0)  # trailing bytes reclaimed
+
+
+def test_compact_pack_detects_overlap(repo_fixtures, request):
+    # Overlapping index entries mean index corruption: compact_pack must raise IntegrityError and
+    # leave the pack in place. Point H(1) back at H(0)'s offset to overlap it.
     chunk0 = fchunk(b"DATA0", chunk_id=H(0))
     chunk1 = fchunk(b"DATA1", chunk_id=H(1))
     repository = get_repository_from_fixture(repo_fixtures, request)
@@ -407,9 +449,26 @@ def test_compact_pack_complete_detects_short_coverage(repo_fixtures, request):
     with repository:
         old_pack_id = repository.chunks[H(0)].pack_id
         entry = repository.chunks[H(1)]
-        repository.chunks[H(1)] = entry._replace(obj_size=entry.obj_size - 1)  # leave 1 trailing byte unaccounted
+        repository.chunks[H(1)] = entry._replace(obj_offset=0)  # now overlaps H(0) at offset 0
 
-        with pytest.raises(AssertionError):
+        with pytest.raises(IntegrityError):
+            repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set())
+        assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")]
+
+
+def test_compact_pack_detects_past_eof(repo_fixtures, request):
+    # An index entry whose span ends past the pack file means index corruption: compact_pack must
+    # raise IntegrityError and leave the pack in place, so defrag never short-reads a truncated object.
+    chunk0 = fchunk(b"DATA0", chunk_id=H(0))
+    chunk1 = fchunk(b"DATA1", chunk_id=H(1))
+    repository = get_repository_from_fixture(repo_fixtures, request)
+    build_one_pack(repository, [(H(0), chunk0), (H(1), chunk1)])
+    with repository:
+        old_pack_id = repository.chunks[H(0)].pack_id
+        entry = repository.chunks[H(1)]
+        repository.chunks[H(1)] = entry._replace(obj_size=entry.obj_size + 1000)  # now claims to end past EOF
+
+        with pytest.raises(IntegrityError):
             repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set())
         assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")]