compact: do not rewrite or merge packs recorded corrupt, refs #10410

docs/internals/data-structures.rst

@@ -227,6 +227,8 @@ Compaction
   - a pack file with some unused objects is rewritten without them, but only if
     the wasted bytes reach the ``--threshold`` percentage
   - very small pack files are merged into bigger ones
+  - a pack file recorded corrupt by ``borg check`` is not rewritten or merged,
+    it is only deleted if its indexed objects are all unused
 - update the chunks index in ``index/`` accordingly
 - with ``--stats``, compute statistics about:
 

docs/internals/packs.rst

@@ -243,6 +243,9 @@ content-addressed name and deletes the old one) and ``borg debug delete-obj``. A
 single blob cannot be removed from a pack in place: all of these paths write a new
 pack file without it and then delete the old one, so store-level deletion always
 operates at pack granularity.
+``borg compact`` (rewriting, merging) and ``borg repo-compress`` skip packs recorded
+corrupt in ``cache/checked-packs``: the rewritten pack would get a new content-addressed
+name that passes ``borg check``, hiding the corruption.
 
 Gap bytes
 ~~~~~~~~~

src/borg/archiver/compact_cmd.py

@@ -12,10 +12,10 @@
 from ..helpers.argparsing import ArgumentParser
 from ..constants import *  # NOQA
 from ..hashindex import ChunkIndex
-from ..helpers import set_ec, EXIT_ERROR, Error, sig_int, format_file_size, bin_to_hex, hex_to_bin, IntegrityError
-from ..helpers import ProgressIndicatorPercent
+from ..helpers import set_ec, EXIT_ERROR, EXIT_WARNING, Error, sig_int, format_file_size, bin_to_hex, hex_to_bin
+from ..helpers import IntegrityError, ProgressIndicatorPercent
 from ..repoobj import object_validator
-from ..repository import Repository
+from ..repository import Repository, PackTracker
 
 from ..logger import create_logger
 
@@ -267,6 +267,10 @@ def compact_packs(self):
         - no indexed objects unused                      -> keep it, unless it is a tiny pack we
                                 merge (see below).
 
+        A pack recorded corrupt in PackTracker is dropped if all its bytes are indexed and unused,
+        otherwise kept unchanged: a pack's id is the hash of its content, so a rewritten or merged copy
+        of the corrupt bytes would get a new id that passes "borg check".
+
         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). compact_pack keeps those; recovering or
         dropping them is "borg check --repair"'s job. See issue #9868.
@@ -327,6 +331,9 @@ def compact_packs(self):
                 "https://github.com/borgbackup/borg/issues/8572."
             )
 
+        # packs recorded corrupt in PackTracker that are still in the store
+        corrupt_packs = set(PackTracker.load(self.repository).corrupt_ids()) & pack_total.keys()
+
         # decide each pack's fate. a pack's reclaimable bytes are its indexed-but-unused bytes; the
         # redundant duplicates compact_pack finds in the gaps are reclaimed on top when it rewrites.
         # a merge fills packs up to pack_max_size, so cap "tiny" at half of that: a merged full pack
@@ -344,13 +351,13 @@ def compact_packs(self):
                 continue  # leave this pack untouched
             reclaimable = indexed - used  # unused indexed bytes; the only bytes compact removes
             if reclaimable == 0:
-                if used == indexed and total < tiny_limit:
+                if used == indexed and total < tiny_limit and pid not in corrupt_packs:
                     merge_packs.add(pid)  # fully-used but tiny -> merge candidate
                 continue  # nothing to reclaim -> leave alone
             if used == 0 and indexed == total:
                 drop_packs.add(pid)  # whole file is unused indexed bytes -> drop it
                 pack_reclaim[pid] = reclaimable
-            elif 100 * reclaimable / total >= self.threshold:
+            elif 100 * reclaimable / total >= self.threshold and pid not in corrupt_packs:
                 rewrite_packs.add(pid)  # wasteful enough -> copy used objects (and unindexed bytes) forward
                 pack_reclaim[pid] = reclaimable
             # else: below threshold -> leave alone
@@ -371,6 +378,16 @@ def compact_packs(self):
         if not worth_merging:
             merge_packs = set()
 
+        kept_corrupt = corrupt_packs - drop_packs
+        if kept_corrupt:
+            logger.warning(
+                f'{len(kept_corrupt)} pack(s) recorded corrupt by "borg check" are not rewritten or merged. '
+                'Run "borg check --repair --verify-data". Damage outside of chunks is not repaired yet, see #10026.'
+            )
+            for pid in sorted(kept_corrupt):
+                logger.debug(f"Corrupt pack: {bin_to_hex(pid)}")
+            set_ec(EXIT_WARNING)
+
         if self.dry_run:
             freed = sum(pack_reclaim.values())
             logger.info(
@@ -488,6 +505,13 @@ def build_parser_compact(self, subparsers, common_parser, mid_common_parser):
             before recording its objects, are re-indexed by ``borg check --repair`` and reclaimed by
             the next ``borg compact``.
 
+            ``borg compact`` does not rewrite or merge packs that ``borg check`` recorded as corrupt
+            and warns about them. ``borg check --repair --verify-data`` deletes the corrupt chunks by
+            rewriting their packs. It does not remove damage outside any chunk (e.g. bytes appended to a
+            pack): a pack with such damage and no corrupt chunk stays recorded corrupt and ``borg compact``
+            warns about it on every run, a pack that also has a corrupt chunk is rewritten with that damage
+            copied into the new pack (refs #10026).
+
             You usually do not want to run ``borg compact`` after every write operation, but
             either regularly (e.g., once a month, possibly together with ``borg check``) or
             when disk space needs to be freed.

src/borg/archiver/repo_compress_cmd.py

@@ -7,11 +7,11 @@
 from ..cache import build_chunkindex_from_repo, delete_chunkindex_from_repo, write_chunkindex_to_repo
 from ..compress import ObfuscateSize, Auto, COMPRESSOR_TABLE
 from ..constants import *  # NOQA
-from ..helpers import sig_int, ProgressIndicatorPercent, Error, CompressionSpec
-from ..helpers import format_file_size, hex_to_bin
+from ..helpers import sig_int, ProgressIndicatorPercent, Error, CompressionSpec, set_ec, EXIT_WARNING
+from ..helpers import format_file_size, bin_to_hex, hex_to_bin
 from ..helpers.argparsing import ArgumentParser
 from ..repoobj import object_validator
-from ..repository import Repository
+from ..repository import Repository, PackTracker
 
 from ..logger import create_logger
 
@@ -56,6 +56,9 @@ class PackRecompressor:
     Each pack is loaded once (via Repository.transform_pack); objects not stored with the desired
     compression get recompressed, the others are carried into the rewritten pack unchanged. A pack
     whose objects all already match is not rewritten at all.
+
+    A pack recorded corrupt in PackTracker is not rewritten: a pack's id is the hash of its content,
+    so a rewritten copy of the corrupt bytes would get a new id that passes "borg check".
     """
 
     def __init__(self, repository, manifest, *, print_stats):
@@ -72,6 +75,7 @@ def __init__(self, repository, manifest, *, print_stats):
         self.objects_recompressed = 0  # recompressed and rewritten
         self.packs_count = 0
         self.packs_rewritten = 0
+        self.packs_corrupt = 0  # packs recorded corrupt, not rewritten
 
     def recompress(self):
         """Recompress the repository pack after pack; Ctrl-C stops cleanly at a pack boundary."""
@@ -91,14 +95,27 @@ def recompress(self):
         size_before = sum(size for _, size in packs)
         size_after = size_before
 
-        stale_packs = set(per_pack) - {pack_id for pack_id, _ in packs}
+        present_packs = {pack_id for pack_id, _ in packs}
+        stale_packs = set(per_pack) - present_packs
         if stale_packs:
             # stale entries reference a pack absent from the store. they are kept, borg check --repair removes them.
             stale = sum(len(per_pack[pack_id]) for pack_id in stale_packs)
             logger.warning(
                 f'index entries referencing a missing pack file: {stale}. Run "borg check --repair" to remove them.'
             )
 
+        # packs recorded corrupt in PackTracker that are still in the store
+        corrupt_packs = set(PackTracker.load(self.repository).corrupt_ids()) & present_packs
+        self.packs_corrupt = len(corrupt_packs)
+        if corrupt_packs:
+            logger.warning(
+                f'{len(corrupt_packs)} pack(s) recorded corrupt by "borg check" are not rewritten. '
+                'Run "borg check --repair --verify-data". Damage outside of chunks is not repaired yet, see #10026.'
+            )
+            for pack_id in sorted(corrupt_packs):
+                logger.debug(f"Corrupt pack: {bin_to_hex(pack_id)}")
+            set_ec(EXIT_WARNING)
+
         pi = ProgressIndicatorPercent(
             total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress"
         )
@@ -108,7 +125,7 @@ def recompress(self):
                 break  # stop cleanly at a pack boundary: save the index below, then raise
             ids = per_pack.get(pack_id)
             # a pack without indexed objects (all-gap) is left for "borg check --repair", see #9868.
-            if ids:
+            if ids and pack_id not in corrupt_packs:
                 new_pack_id, new_size = self.repository.transform_pack(
                     pack_id,
                     ids,
@@ -171,7 +188,9 @@ def report(self, size_before, size_after):
         stats_logger = logging.getLogger("borg.output.stats")
         objects = self.objects_ok + self.objects_kept + self.objects_recompressed
         stats_logger.info("Recompression stats:")
-        stats_logger.info(f"Packs: {self.packs_count} total, {self.packs_rewritten} rewritten.")
+        # the objects of skipped corrupt packs are not read, so they are not in the objects total.
+        corrupt = f", {self.packs_corrupt} skipped (recorded corrupt)" if self.packs_corrupt else ""
+        stats_logger.info(f"Packs: {self.packs_count} total, {self.packs_rewritten} rewritten{corrupt}.")
         stats_logger.info(
             f"Objects: {objects} total, {self.objects_recompressed} recompressed, "
             f"{self.objects_ok} already had the desired compression, "
@@ -214,6 +233,13 @@ def build_parser_repo_compress(self, subparsers, common_parser, mid_common_parse
         desired compression type/level - if no compression gives a shorter output, that
         might be chosen; such chunks are kept as they are.
 
+        ``borg repo-compress`` does not rewrite packs that ``borg check`` recorded as corrupt
+        and warns about them. ``borg check --repair --verify-data`` deletes the corrupt chunks by
+        rewriting their packs. It does not remove damage outside any chunk (e.g. bytes appended to a
+        pack): a pack with such damage and no corrupt chunk stays recorded corrupt and is not rewritten,
+        a pack that also has a corrupt chunk is rewritten with that damage copied into the new pack
+        (refs #10026).
+
         Rewriting a pack invalidates every client's cached chunk index, so the next borg
         operation of each client will re-fetch the chunk index from the repository.
 

src/borg/testsuite/archiver/compact_cmd_test.py

@@ -1,3 +1,4 @@
+import logging
 import os
 from pathlib import Path
 from types import SimpleNamespace
@@ -8,7 +9,7 @@
 from ...helpers import get_cache_dir, bin_to_hex, sig_int, Error
 from ...hashindex import ChunkIndex
 from ...repoobj import RepoObj
-from ...repository import Repository
+from ...repository import Repository, PackTracker
 from ...cache import files_cache_name, discover_files_cache_names, list_chunkindex_hashes
 from ...cache import delete_chunkindex_from_repo, write_chunkindex_to_repo
 from ...manifest import Manifest
@@ -19,7 +20,7 @@
 from . import cmd, create_regular_file, create_src_archive, generate_archiver_tests, open_repository, RK_ENCRYPTION
 from . import changedir
 from .. import make_test_key
-from ..repository_test import H, fchunk, pdchunk
+from ..repository_test import H, fchunk, pdchunk, corrupt_chunk_on_disk
 
 pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary")  # NOQA
 
@@ -444,6 +445,43 @@ def test_compact_keeps_undelete_data_when_chunks_missing(archivers, request):
         assert fd.read() == b"G" * (1024 * 80)  # its data survived compaction
 
 
+def test_compact_keeps_corrupt_pack(archivers, request):
+    # compact keeps a pack "borg check" recorded corrupt unchanged and warns (#10410).
+    archiver = request.getfixturevalue(archivers)
+
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    keep = os.urandom(200_000)
+    create_regular_file(archiver.input_path, "s3/keep", contents=keep)
+    create_regular_file(archiver.input_path, "s3/drop", contents=os.urandom(30_000))
+    create_regular_file(archiver.input_path, "s3k/keep", contents=keep)
+    cmd(archiver, "create", "a", "input/s3")
+    cmd(archiver, "create", "b", "input/s3k")  # dedups against a
+    cmd(archiver, "delete", "-a", "a")  # the s3/drop chunk is unused now
+    repository = open_repository(archiver)
+    with repository:
+        manifest = Manifest.load(repository)
+        b = Archive(manifest, manifest.archives.get_one(["b"]).id)
+        keep_id = next(id for item in b.iter_items() if "chunks" in item for id, _ in item.chunks)
+        pack_key = "packs/" + bin_to_hex(repository.chunks[keep_id].pack_id)
+        corrupt_chunk_on_disk(repository, keep_id)
+        pack_before = repository.store_load(pack_key)
+    cmd(archiver, "check", exit_code=1)
+
+    output = cmd(archiver, "compact", "-v", "--threshold", "0", exit_code=EXIT_WARNING)
+    assert '1 pack(s) recorded corrupt by "borg check" are not rewritten or merged.' in output
+    assert "Damage outside of chunks is not repaired yet, see #10026." in output
+    repository = open_repository(archiver)
+    with repository:
+        assert repository.store_load(pack_key) == pack_before
+    cmd(archiver, "check", exit_code=1)
+
+    # the repair deletes the corrupt chunk: compact reports it missing and no corrupt pack.
+    cmd(archiver, "check", "--repair", "--verify-data", exit_code=EXIT_SUCCESS)
+    output = cmd(archiver, "compact", "-v", "--threshold", "0", exit_code=EXIT_ERROR)
+    assert "missing objects" in output
+    assert "recorded corrupt" not in output
+
+
 def test_compact_keeps_stale_index_entries(tmp_path):
     # A stale index entry references a pack file absent from the store. compact keeps and reports it; a
     # used stale entry means data is missing (#9850).
@@ -596,6 +634,116 @@ def test_compact_packs_below_all_packs_gate_changes_nothing(tmp_path):
         assert pdchunk(repository.get(H(1))) == b"x"
 
 
+def record_corrupt(repository, pack_id):
+    """Record pack_id as corrupt in PackTracker."""
+    tracker = PackTracker.load(repository)
+    tracker.record(pack_id, False)
+    tracker.save()
+
+
+@pytest.mark.parametrize("dry_run", (False, True))
+def test_compact_packs_does_not_rewrite_corrupt_pack(tmp_path, caplog, dry_run):
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        manifest = gc_manifest(repository)
+        repository._pack_writer.max_count = 4
+        for i in range(3):
+            repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i)))
+        repository.flush()
+        pack_id = repository.chunks[H(0)].pack_id
+        for i in range(3):  # H0 used, H1 and H2 unused
+            flags = ChunkIndex.F_USED if i == 0 else ChunkIndex.F_NONE
+            repository.chunks[H(i)] = repository.chunks[H(i)]._replace(flags=flags)
+        record_corrupt(repository, pack_id)
+
+        gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=0, dry_run=dry_run)
+        gc.chunks = repository.chunks
+        with caplog.at_level(logging.WARNING):
+            gc.compact_packs()
+
+        assert '1 pack(s) recorded corrupt by "borg check" are not rewritten or merged.' in caplog.text
+        assert gc.store_changed is False
+        assert [info.name for info in repository.store_list("packs")] == [bin_to_hex(pack_id)]
+        for i in range(3):
+            assert repository.chunks[H(i)].pack_id == pack_id
+
+
+def test_compact_packs_does_not_merge_corrupt_pack(tmp_path, caplog, monkeypatch):
+    monkeypatch.setenv("BORG_PACK_MAX_SIZE", "300")
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        manifest = gc_manifest(repository)
+        num = 10
+        for i in range(num):
+            repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i)))
+            repository.flush()  # one tiny pack per object
+        for i in range(num):
+            repository.chunks[H(i)] = repository.chunks[H(i)]._replace(flags=ChunkIndex.F_USED)
+        corrupt_pack = repository.chunks[H(0)].pack_id
+        record_corrupt(repository, corrupt_pack)
+
+        gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10)
+        gc.chunks = repository.chunks
+        with caplog.at_level(logging.WARNING):
+            gc.compact_packs()
+
+        assert gc.store_changed is True  # the other tiny packs were merged
+        assert "recorded corrupt" in caplog.text
+        assert bin_to_hex(corrupt_pack) in {info.name for info in repository.store_list("packs")}
+        assert repository.chunks[H(0)].pack_id == corrupt_pack
+        for i in range(1, num):
+            assert repository.chunks[H(i)].pack_id != corrupt_pack
+
+
+def test_compact_packs_corrupt_pack_does_not_count_toward_merge_gate(tmp_path, monkeypatch):
+    # the tiny packs reach a full pack only together with the corrupt one, so nothing is merged.
+    monkeypatch.setenv("BORG_PACK_MAX_SIZE", "300")
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        manifest = gc_manifest(repository)
+        num = 0
+        total = 0
+        while total < repository.pack_max_size:
+            repository.put(H(num), fchunk(f"DATA{num}".encode(), chunk_id=H(num)))
+            repository.flush()  # one tiny pack per object
+            repository.chunks[H(num)] = repository.chunks[H(num)]._replace(flags=ChunkIndex.F_USED)
+            total += repository.store.info("packs/" + bin_to_hex(repository.chunks[H(num)].pack_id)).size
+            num += 1
+        record_corrupt(repository, repository.chunks[H(num - 1)].pack_id)
+        packs_before = {info.name for info in repository.store_list("packs")}
+
+        gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10)
+        gc.chunks = repository.chunks
+        gc.compact_packs()
+
+        assert gc.store_changed is False
+        assert {info.name for info in repository.store_list("packs")} == packs_before
+
+
+def test_compact_packs_drops_unused_corrupt_pack(tmp_path, caplog):
+    # a corrupt pack whose bytes are all indexed and unused is dropped.
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True) as repository:
+        manifest = gc_manifest(repository)
+        repository.put(H(0), fchunk(b"DATA0", chunk_id=H(0)))
+        repository.flush()
+        pack_id = repository.chunks[H(0)].pack_id
+        repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_NONE)
+        record_corrupt(repository, pack_id)
+
+        gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10)
+        gc.chunks = repository.chunks
+        with caplog.at_level(logging.WARNING):
+            gc.compact_packs()
+
+        assert "recorded corrupt" not in caplog.text
+        assert gc.store_changed is True
+        assert list(repository.store_list("packs")) == []
+        assert H(0) not in gc.chunks
+
+
 def test_compact_gc_after_index_loss(archivers, request):
     """When no 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),