index rebuild: abort cleanly on a corrupt object header, #10122

docs/internals/frontends.rst

@@ -923,6 +923,8 @@ Errors
         File failed integrity check: {}
     DecompressionError rc: 92 traceback: yes
         Decompression error: {}
+    CorruptPack rc: 93 traceback: no
+        {}. Run "borg check --repair" to recover the objects that are still readable.
 
     Reading a legacy borg 1.x repository (e.g. ``borg transfer --from-borg1``) raises the
     ``LegacyRepository.*`` and ``LegacyRemoteRepository.*`` variants of the repository and RPC

docs/internals/packs.rst

@@ -100,7 +100,10 @@ A reader locates the next blob by advancing::
 ``iter_headers()`` checks every header it walks: it must have ``OBJ_MAGIC``, a
 supported version, and sizes that keep the blob inside the pack and within
 ``MAX_DATA_SIZE``. A header that fails these checks means a corrupt pack, and
-``IntegrityError`` is raised, naming which check it failed.
+``IntegrityError`` is raised, naming which check it failed. A chunks index
+rebuild without a repair walk (see below) would be incomplete from that point
+on, so it turns that into ``CorruptPack``, telling the user to run
+``borg check --repair``.
 
 The per-blob magic limits the blast radius of corrupted length fields. The
 repair walk (``iter_headers(validate=...)``, used when ``borg check --repair``

src/borg/archive.py

@@ -2224,6 +2224,12 @@ def check(
             drop_corrupt_tail=repair,
             write_immediately=False,
         )
+        if repair:
+            # repository.chunks is a separate index, lazily built when repository.get() resolves a
+            # chunk location. It walks the same packs, so give it the same corrupt-header handling
+            # the rebuild above got - otherwise the repair aborts at a header it just resynced past.
+            self.repository.chunkindex_validate = validate
+            self.repository.chunkindex_drop_corrupt_tail = True
         if self.key is None:
             self.key = self.make_key(repository)
             self.repo_objs = RepoObj(self.key)

src/borg/cache.py

@@ -29,6 +29,7 @@
 from .helpers import get_cache_dir
 from .helpers import archive_hostname, archive_username
 from .helpers import chunkit
+from .helpers import CorruptPack, IntegrityError
 from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list
 from .helpers import format_file_size, safe_encode
 from .helpers import safe_ns
@@ -909,6 +910,8 @@ def build_chunkindex_from_repo(
     # the walk skips content. It only reports, it does not change what the walk does.
     # drop_corrupt_tail: without a validator, index a pack with a corrupt object header up to that
     # header and drop the rest of it, instead of raising, see PackReader.iter_headers.
+    # With neither of the two, a corrupt object header aborts the rebuild with CorruptPack: the
+    # index would be missing every object after it.
     assert not (slow_rebuild and fragments_only)
     assert not (fragments_only and write_immediately)  # fragments_only never writes to the repo
     # first, try to build a fresh, mostly complete chunk index from centrally stored index fragments:
@@ -1004,13 +1007,18 @@ def build_chunkindex_from_repo(
         pi.show(increase=1)
         pack_id = hex_to_bin(info.name)
         reader = PackReader(repository.store, pack_id)
-        for chunk_id, obj_offset, obj_size in reader.iter_headers(
-            validate=validate, on_drop=on_drop, drop_corrupt_tail=drop_corrupt_tail
-        ):
-            num_chunks += 1
-            chunks[chunk_id] = ChunkIndexEntry(
-                flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size
-            )
+        try:
+            for chunk_id, obj_offset, obj_size in reader.iter_headers(
+                validate=validate, on_drop=on_drop, drop_corrupt_tail=drop_corrupt_tail
+            ):
+                num_chunks += 1
+                chunks[chunk_id] = ChunkIndexEntry(
+                    flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size
+                )
+        except IntegrityError as err:
+            # the walk stopped at a corrupt object header, so this index would be incomplete: abort
+            # and point at "borg check --repair", which resyncs past the damage.
+            raise CorruptPack(err) from err
         headers_parsed += reader.headers_parsed
     if pack_infos:
         pi.show(current=len(pack_infos))  # finish at 100%

src/borg/helpers/__init__.py

@@ -12,7 +12,8 @@
 from ..constants import *  # NOQA
 
 from .datastruct import StableDict, Buffer, EfficientCollectionQueue
-from .errors import Error, ErrorWithTraceback, IntegrityError, DecompressionError, CancelledByUser, CommandError
+from .errors import Error, ErrorWithTraceback, IntegrityError, DecompressionError, CorruptPack
+from .errors import CancelledByUser, CommandError
 from .errors import RTError, PathNotAllowed, modern_ec
 from .errors import BorgWarning, FileChangedWarning, BackupWarning, IncludePatternNeverMatchedWarning
 from .errors import BackupError, BackupOSError, BackupRaceConditionError, BackupItemExcluded

src/borg/helpers/errors.py

@@ -60,6 +60,12 @@ class DecompressionError(IntegrityError):
     exit_mcode = 92
 
 
+class CorruptPack(Error):
+    """{}. Run "borg check --repair" to recover the objects that are still readable."""
+
+    exit_mcode = 93
+
+
 class CancelledByUser(Error):
     """Cancelled by user."""
 

src/borg/repository.py

@@ -495,9 +495,9 @@ def iter_headers(self, validate=None, on_drop=None, drop_corrupt_tail=False):
                     # no validator, so payload bytes that look like a header can not be told from
                     # an object: there is no way to resync past this header.
                     if not drop_corrupt_tail:
-                        raise IntegrityError(
-                            f'pack {pack_hex}: {problem} at offset {offset} (pack corruption), run "borg check"'
-                        )
+                        # the callers that can say something more useful than "there is corruption
+                        # here" wrap this, see build_chunkindex_from_repo.
+                        raise IntegrityError(f"pack {pack_hex}: {problem} at offset {offset} (pack corruption)")
                     if on_drop is not None:
                         on_drop()
                     logger.warning(
@@ -872,6 +872,12 @@ def __init__(
         self.exclusive = exclusive
         self._pack_writer = None
         self._chunks = None  # ChunkIndex; loaded lazily on first access to .chunks
+        # corrupt-header handling for the lazy .chunks rebuild, set by ArchiveChecker.check() when
+        # repairing (see PackReader.iter_headers): a validate callable makes the rebuild resync past
+        # a corrupt object header, drop_corrupt_tail makes it index the pack up to that header and
+        # drop the rest. Without either, such a header aborts the rebuild.
+        self.chunkindex_validate = None
+        self.chunkindex_drop_corrupt_tail = False
         # pack_id -> PackReader holding the whole pack; get_many loads into it, get() reuses it
         self._pack_cache = LRUCache(capacity=self.PACK_READER_CACHE_SIZE)
 
@@ -1039,7 +1045,9 @@ def chunks(self):
         if self._chunks is None:
             from .cache import build_chunkindex_from_repo
 
-            self._chunks = build_chunkindex_from_repo(self)
+            self._chunks = build_chunkindex_from_repo(
+                self, validate=self.chunkindex_validate, drop_corrupt_tail=self.chunkindex_drop_corrupt_tail
+            )
         return self._chunks
 
     @chunks.setter

src/borg/testsuite/archiver/check_cmd_test.py

@@ -11,7 +11,7 @@
 from ...archive import ArchiveChecker, ChunkBuffer
 from ...cache import delete_chunkindex_from_repo
 from ...constants import *  # NOQA
-from ...helpers import bin_to_hex, msgpack, CommandError, Error, IntegrityError, sig_int
+from ...helpers import bin_to_hex, msgpack, CommandError, CorruptPack, Error, IntegrityError, sig_int
 from ...manifest import Archives, Manifest
 from ...repoobj import RepoObj
 from ...repository import PackTracker, Repository
@@ -893,15 +893,46 @@ def build_chunkindex_from_repo(repository, **kwargs):
     monkeypatch.setattr(archive_module, "build_chunkindex_from_repo", build_chunkindex_from_repo)
     # --archives-only: the repository check would stop at the damaged pack (a pack is named by the
     # sha256 of its content) before the archives check ever walks it.
-    with pytest.raises(IntegrityError) as excinfo:
+    with pytest.raises(CorruptPack) as excinfo:
         cmd(archiver, "check", "--archives-only")
     assert f"no object header at offset {damaged_offset} (pack corruption)" in str(excinfo.value)
     drop_corrupt_tail, outcome = rebuilds[0]
     # the rebuild raised, it did not return an index with the pack's tail missing
-    assert isinstance(outcome, IntegrityError)
+    assert isinstance(outcome, CorruptPack)
     assert drop_corrupt_tail is False  # a check that only diagnoses does not ask for the drop
 
 
+def test_repo_list_aborts_cleanly_on_corrupt_pack(archivers, request):
+    """A command rebuilding the chunks index over a corrupt object header aborts with a hint (#10122)."""
+    archiver = request.getfixturevalue(archivers)
+    if archiver.get_kind() != "local":
+        pytest.skip("inspects the store directly")
+    check_cmd_setup(archiver)
+    cmd(archiver, "check", exit_code=0)
+
+    with Repository(archiver.repository_location, exclusive=True) as repository:
+        # damage the header of the 2nd object of a pack holding more than 2, so the walk aborts mid-pack.
+        by_pack = {}
+        for entry in repository.chunks.values():
+            by_pack.setdefault(entry.pack_id, []).append(entry.obj_offset)
+        pack_id, offsets = next((p, sorted(o)) for p, o in by_pack.items() if len(o) > 2)
+        damaged_offset = offsets[1]
+        key = "packs/" + bin_to_hex(pack_id)
+        repository.store_store(key, corrupt(repository.store_load(key), damaged_offset))
+        # drop the index fragments, so the next command has to read the pack headers.
+        delete_chunkindex_from_repo(repository)
+
+    # fork: only a subprocess runs borg's top-level error handler, which turns the Error into a rc.
+    output = cmd(archiver, "repo-list", fork=True, exit_code=CorruptPack.exit_mcode)
+    assert "Traceback" not in output
+    assert f"no object header at offset {damaged_offset} (pack corruption)" in output
+    assert "borg check --repair" in output
+
+    # --repair passes a validator, so it resyncs past the damaged header instead of aborting.
+    # TODO: it does not rewrite the pack yet, so a later rebuild hits the same header again.
+    cmd(archiver, "check", "--repair", exit_code=0)
+
+
 def test_repair_finish_flushes_pack_writer(archivers, request):
     """finish() stores chunks re-added during --repair before it (re)builds the index (#10055).
 

src/borg/testsuite/cache_test.py

@@ -26,7 +26,7 @@
 )
 from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..crypto.key import AESOCBKey
-from ..helpers import Error, IntegrityError, bin_to_hex, safe_ns
+from ..helpers import CorruptPack, Error, bin_to_hex, safe_ns
 from ..helpers.msgpack import int_to_timestamp
 from ..manifest import Manifest
 from ..repository import Repository
@@ -506,7 +506,7 @@ def test_close_consolidates_fragments_across_sessions(tmp_path, monkeypatch):
 
 
 def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path):
-    """A corrupt object header fails the rebuild; with validate, the objects after it are indexed."""
+    """A corrupt object header aborts the rebuild; with validate, the objects after it are indexed."""
     from .repository_test import accept_all, fchunk
 
     obj1 = bytearray(fchunk(b"first", chunk_id=H(90)))
@@ -515,7 +515,7 @@ def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path):
     pack_id = H(92)
     with Repository(os.fspath(tmp_path / "repository"), exclusive=True, create=True) as repository:
         repository.store_store("packs/" + bin_to_hex(pack_id), bytes(obj1) + obj2)
-        with pytest.raises(IntegrityError):
+        with pytest.raises(CorruptPack, match="borg check --repair"):  # no validator: abort, #10122
             build_chunkindex_from_repo(repository, slow_rebuild=True)
         # accept_all takes any candidate, so this covers the plumbing, not the authentication.
         index = build_chunkindex_from_repo(repository, slow_rebuild=True, validate=accept_all)
@@ -577,7 +577,7 @@ def test_build_chunkindex_without_drop_corrupt_tail_raises_on_a_damaged_pack(tmp
     drops = []
     with Repository(os.fspath(tmp_path / "repository"), exclusive=True, create=True) as repository:
         repository.store_store("packs/" + bin_to_hex(H(93)), obj1 + bytes(obj2))
-        with pytest.raises(IntegrityError, match="no object header at offset"):
+        with pytest.raises(CorruptPack, match="no object header at offset"):
             build_chunkindex_from_repo(repository, slow_rebuild=True, on_drop=lambda: drops.append(True))
         assert drops == []  # nothing was discarded: the walk did not get that far