diff --git a/docs/internals/data-structures.rst b/docs/internals/data-structures.rst index 42750081ba..b4bce67abb 100644 --- a/docs/internals/data-structures.rst +++ b/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: diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 8d09cdeac9..574c1ae7b8 100644 --- a/docs/internals/packs.rst +++ b/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 ~~~~~~~~~ diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 05f8914a35..e281fd5874 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/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. diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py index 66f759c92b..23b331dc2a 100644 --- a/src/borg/archiver/repo_compress_cmd.py +++ b/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,7 +95,8 @@ 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) @@ -99,6 +104,18 @@ def recompress(self): 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. diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index 13368fe42d..58e670b2ed 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/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), diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 140473303e..9a12141909 100644 --- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py +++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py @@ -13,7 +13,7 @@ from .. import make_test_key from . import create_regular_file, cmd, open_repository, RK_ENCRYPTION -from ..repository_test import H, accept_all, fchunk, pdchunk +from ..repository_test import H, accept_all, fchunk, pdchunk, corrupt_chunk_on_disk def test_repo_compress(archiver): @@ -240,6 +240,41 @@ def store_delete_then_interrupt(name, **kwargs): cmd(archiver, "check") +@pytest.mark.parametrize("corrupt_ctype", (CNONE.ID, ZLIB.ID)) +def test_repo_compress_keeps_corrupt_pack(archiver, corrupt_ctype): + # repo-compress keeps a pack "borg check" recorded corrupt unchanged and warns (#10410). + # with "-C none", the corrupt object is either already stored as wanted (CNONE) or needs recompressing (ZLIB). + create_regular_file(archiver.input_path, "rand", contents=os.urandom(100_000)) # stored uncompressed + create_regular_file(archiver.input_path, "text", contents=b"z" * 100_000) # stored zlib compressed + cmd(archiver, "repo-create", RK_ENCRYPTION) + cmd(archiver, "create", "test", "input", "-C", "zlib,3") + with open_repository(archiver) as repository: + manifest = Manifest.load(repository) + ctypes = {} # pack_id -> {ctype, ...} + for id, entry in repository.chunks.iteritems(): + meta = manifest.repo_objs.parse_meta(id, repository.get(id, read_data=False), ro_type=ROBJ_DONTCARE) + ctypes.setdefault(entry.pack_id, set()).add(meta["ctype"]) + if meta["ctype"] == corrupt_ctype and meta["size"] == 100_000: + corrupt_id = id + corrupt_pack = repository.chunks[corrupt_id].pack_id + # the corrupt pack holds both objects, and another pack holds an object "-C none" recompresses. + assert ctypes.pop(corrupt_pack) >= {CNONE.ID, ZLIB.ID} + assert any(ZLIB.ID in pack_ctypes for pack_ctypes in ctypes.values()) + packs_before = {info.name for info in repository.store_list("packs")} + corrupt_chunk_on_disk(repository, corrupt_id) # corrupts the data, the metadata stays readable + cmd(archiver, "check", exit_code=1) + + output = cmd(archiver, "repo-compress", "-v", "--stats", "-C", "none", exit_code=EXIT_WARNING) + assert '1 pack(s) recorded corrupt by "borg check" are not rewritten.' in output + assert "Damage outside of chunks is not repaired yet, see #10026." in output + assert "rewritten, 1 skipped (recorded corrupt)." in output + with open_repository(archiver) as repository: + packs_after = {info.name for info in repository.store_list("packs")} + assert bin_to_hex(corrupt_pack) in packs_after + assert len(packs_before - packs_after) >= 1 # other packs were recompressed + cmd(archiver, "check", exit_code=1) + + def transform_via(replacements): """Make a transform_pack callback replacing the given objects, keeping all others unchanged."""