diff --git a/src/borg/archive.py b/src/borg/archive.py index 1a09317492..939d2f86ff 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -23,7 +23,7 @@ from . import xattr from .chunkers import get_chunker, Chunk, release_chunk_data -from .cache import ChunkListEntry, build_chunkindex_from_repo, delete_chunkindex_from_repo +from .cache import ChunkListEntry, build_chunkindex_from_repo, write_chunkindex_to_repo from .crypto.key import key_factory, UnsupportedPayloadError from .constants import * # NOQA from .crypto.low_level import IntegrityError as IntegrityErrorBase @@ -1863,6 +1863,9 @@ class ArchiveChecker: def __init__(self): self.error_found = False self.key = None + # True once repair drops a defect chunk or writes a new one, i.e. once the chunks index no + # longer matches the packs. + self.chunks_modified = False def check( self, @@ -2060,6 +2063,7 @@ def verify_data(self): # keeping the other chunks. update_index=False: finish() rebuilds the index from # the rewritten packs anyway, so a per-chunk full index write would be wasted. self.repository.delete(defect_chunk, update_index=False) + self.chunks_modified = True # drop it from our own index too, so rebuild_archives reports the file it belongs to. del self.chunks[defect_chunk] else: @@ -2216,6 +2220,7 @@ def add_reference(id_, size, cdata): if self.repair: pack_results = self.repository.put(id_, cdata) self.chunks.update_pack_info(pack_results) + self.chunks_modified = True def verify_file_chunks(archive_name, item): """Verify that all of a file's chunks are present, collecting any missing ones for the report.""" @@ -2429,10 +2434,20 @@ def valid_item(obj): def finish(self): if self.repair: - # we may have deleted chunks. delete_chunkindex_from_repo() removes the on-disk index and - # drops the stale in-memory index, so the next repository access rebuilds it from the repo. - logger.info("Deleting chunk indexes in repository - next repository access will cause a rebuild.") - delete_chunkindex_from_repo(self.repository) + if self.chunks_modified: + # the packs changed, so the index no longer matches them: rebuild it from the packs + # and persist it. flush first so the rewritten and newly written packs are on the store. + self.repository.flush() + logger.info("Rebuilding and writing the repository chunks index.") + build_chunkindex_from_repo(self.repository, slow_rebuild=True, write_immediately=True) + else: + # the packs are unchanged, so the index still matches them: persist it as is. + logger.info("Writing the rebuilt repository chunks index.") + write_chunkindex_to_repo( + self.repository, self.chunks, incremental=False, clear=False, force_write=True, delete_other=True + ) + # drop the in-memory index so close() does not persist it over the index just written. + self.repository.invalidate_chunk_index() logger.info("Writing Manifest.") self.manifest.write() diff --git a/src/borg/archiver/check_cmd.py b/src/borg/archiver/check_cmd.py index fa31c6a98b..717051fcb1 100644 --- a/src/borg/archiver/check_cmd.py +++ b/src/borg/archiver/check_cmd.py @@ -78,7 +78,9 @@ def do_check(self, args, repository): # the repository check has finished, which can take hours. ArchiveFormatter.validate_format(format) if not args.archives_only: - if not repository.check(repair=args.repair, max_duration=args.max_duration, max_age=max_age): + if not repository.check( + repair=args.repair, max_duration=args.max_duration, max_age=max_age, repo_only=args.repo_only + ): set_ec(EXIT_WARNING) if sig_int: # repository check interrupted; skip the archive check raise Error("Got Ctrl-C / SIGINT.") @@ -232,8 +234,10 @@ def build_parser_check(self, subparsers, common_parser, mid_common_parser): In practice, repair mode hooks into both the repository and archive checks: - 1. When checking the repository's consistency, repair mode removes corrupted - objects from the repository after it did a 2nd try to read them correctly. + 1. When checking the repository's consistency, repair mode rebuilds the repository + index from the packs if the index is corrupt, provided every pack is intact. If + any pack is corrupt, the index is left as-is and the corruption is reported; + salvaging a corrupt pack's still-intact objects is not implemented yet. 2. When checking the consistency and correctness of archives, repair mode might remove whole archives from the manifest if their archive metadata chunk is diff --git a/src/borg/cache.py b/src/borg/cache.py index 0f539adbb3..4b76085d47 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -30,7 +30,7 @@ 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 -from .helpers import ProgressIndicatorMessage +from .helpers import ProgressIndicatorMessage, ProgressIndicatorPercent from .helpers import msgpack from .helpers.msgpack import int_to_timestamp, timestamp_to_int from .item import ChunkListEntry @@ -873,15 +873,23 @@ def build_chunkindex_from_repo( # headers and skipping the (much larger) encrypted payloads. Don't call Repository.list() here: # it iterates this same index we are building, so it would recurse. The headers also give each # object's real (chunk_id, offset, size), so every object in a pack is indexed individually. - for info in repository.store_list("packs"): + pack_infos = repository.store_list("packs") + pi = ProgressIndicatorPercent( + total=len(pack_infos), msg="Rebuilding chunk index %3.0f%%", msgid="cache.build_chunkindex_from_repo" + ) + for info in pack_infos: # PackReader uses the store directly, so refresh the lock here; a full rebuild can be slow. repository._lock_refresh() + pi.show(increase=1) pack_id = hex_to_bin(info.name) for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(): num_chunks += 1 chunks[chunk_id] = ChunkIndexEntry( flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size ) + if pack_infos: + pi.show(current=len(pack_infos)) # finish at 100% + pi.finish() duration = perf_counter() - t0 or 0.001 # Chunk IDs in a list are encoded in 34 bytes: 1 byte msgpack header, 1 byte length, 32 ID bytes. # Protocol overhead is neglected in this calculation. diff --git a/src/borg/repository.py b/src/borg/repository.py index 120a0f59ec..66edaf7671 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -967,7 +967,7 @@ def info(self): info = dict(id=self.id, version=self.version) return info - def check(self, repair=False, max_duration=0, max_age=0): + def check(self, repair=False, max_duration=0, max_age=0, repo_only=False): """Check repository consistency. packs/ and index/ objects are named by the sha256 of their content, so a pack or index file @@ -977,18 +977,26 @@ def check(self, repair=False, max_duration=0, max_age=0): The index is hashed first and the packs only if it is intact. The packs could be hashed even with a corrupt index, but a corrupt index already means the user has to repair it, and that rebuild re-reads every pack anyway - so a read-only check just stops and reports it instead of - continuing. The index is never rebuilt here in any case: reading every pack to do so would be - far too slow and expensive for a routine (e.g. cron) check. Salvaging good objects out of - corrupt packs and dropping those packs is left to repair, refs #8572. The ids of the packs - found corrupt are kept in cache/checked-packs for repair, refs #9696. + continuing. A read-only check never rebuilds the index: reading every pack to do so would be + far too slow and expensive for a routine (e.g. cron) check. With repair=True and a corrupt + index, and if every pack is intact, the index is rebuilt from the packs' object headers and + persisted; on a full check the archives phase rebuilds and re-persists it afterwards, see + ArchiveChecker.finish. Packs are verified by sha256, which is content-addressing rather than a + MAC, so this rebuild detects accidental corruption but not tampering, refs #9901, #10026. If any + pack is corrupt the index is left unchanged, refs #8572, #10026. Pack ids found corrupt are kept + in cache/checked-packs, refs #9696. A pack recorded corrupt fails the check, also on a partial run that stops before re-reaching it. The record clears at the check that finds the pack intact again or gone (removed by - compact, or salvaged and dropped by repair; refs #8572); prune() does this from packs/. + compact; TODO: also when repair salvages and drops it, refs #8572); prune() does this from packs/. max_age (seconds, 0 = verify every pack): skip packs whose intact record is younger than max_age, accepting a future timestamp up to MAX_CLOCK_SKEW (clock skew). Results are recorded regardless of max_age. + + repo_only: whether this is a repository-only run. In repair mode it sets the return value for a + corrupt pack, which repair does not fix: fail if repo_only, else defer (a full check's archives + phase can repair a corrupt pack holding metadata, or file content with --verify-data). """ def verify(namespace, name): @@ -1027,6 +1035,8 @@ def store_list(namespace): t_last_checkpoint = t_start index_files = index_errors = 0 pack_files = pack_errors = pack_skipped = 0 + index_repaired = False + packs_scanned = False # index and packs get separate progress indicators, each running from 0% to 100%. # the index is checked first and in full, on partial checks too: it is small, and index errors # stop the pack check below. @@ -1047,7 +1057,13 @@ def store_list(namespace): if index_infos: index_pi.show(current=len(index_infos)) # finish at 100% index_pi.finish() - if index_errors == 0: + if index_errors == 0 or repair: + # verify the packs; during repair, rebuild the corrupt index from them afterwards. + # --repair forbids --max-duration and --max-age, so the partial and max_age handling in + # the loop stays inactive during a repair. + packs_scanned = True + if index_errors: + logger.warning("Repository index is corrupted; rebuilding it from the packs.") # packs are the bulk of the work and the part --max-duration spreads over several checks. pack_infos = store_list("packs") # drop objects whose name is not a valid pack name and count them as errors; the code @@ -1106,8 +1122,16 @@ def recorded_ts(info): logger.info("Finished checking packs.") tracker.prune({hex_to_bin(info.name) for info in pack_infos}) pack_pi.finish() + if index_errors and pack_errors == 0: + from .cache import build_chunkindex_from_repo + + # rebuild the index from the packs. the exclusive check lock keeps the pack set + # fixed, so re-listing packs/ inside build_chunkindex_from_repo matches this + # verification. write_immediately persists the index and drops the corrupt fragments. + build_chunkindex_from_repo(self, slow_rebuild=True, write_immediately=True) + self.invalidate_chunk_index() # the rebuilt index is persisted; drop the in-memory copy + index_repaired = True else: - # TODO: --repair will rebuild the index from the packs here instead of stopping (refs #8572). logger.error("Repository index is corrupted and must be repaired; skipping the pack check.") objs_errors = index_errors + pack_errors summary = ( @@ -1117,11 +1141,13 @@ def recorded_ts(info): if pack_skipped: summary += f" Reused {pack_skipped} recent pack check result(s)." logger.info(summary) - # corrupt_ids() is every pack recorded corrupt, including from earlier runs. with a corrupt - # index the packs were not scanned, so report nothing. - corrupt_ids = tracker.corrupt_ids() if index_errors == 0 else [] + if index_repaired: + logger.info("Repository index was corrupted and has been rebuilt from the packs.") + # corrupt_ids() includes packs recorded corrupt in earlier runs; report them only when this + # run scanned the packs. + corrupt_ids = tracker.corrupt_ids() if packs_scanned else [] if corrupt_ids: - # one id per line (the list can be long). + # one id per line, the list can be long. logger.error(f"Found {len(corrupt_ids)} corrupt pack(s):") for pack_id in corrupt_ids: logger.error(f"Corrupt pack: {bin_to_hex(pack_id)}") @@ -1131,12 +1157,25 @@ def recorded_ts(info): done, so_far = ("Interrupted", " so far") if sig_int else ("Finished", "") if not problems: logger.info(f"{done} {mode} repository check, no problems found{so_far}.") - elif repair: - logger.error(f"{done} {mode} repository check, errors found{so_far} (repository repair not implemented).") - else: + elif not repair: logger.error(f"{done} {mode} repository check, errors found{so_far}.") - # True means the checked objects were clean; --repair returns True so the caller proceeds to fix them. - return not problems or repair + elif not (pack_errors or corrupt_ids): + # only the index was corrupt, and it was rebuilt. + logger.info(f"{done} {mode} repository check, repaired{so_far}.") + elif repo_only: + logger.error( + f"{done} {mode} repository check, corrupt pack(s) found{so_far}; repairing a repository " + "with corrupt packs is not implemented yet (refs #8572)." + ) + else: + # a full check's archives phase reads archive/item metadata (and file content with + # --verify-data), so it repairs a corrupt pack holding such objects; warn rather than fail. + logger.warning(f"{done} {mode} repository check, corrupt pack(s) found{so_far}.") + # in repair mode a corrupt pack fails only a repository-only run; a full check defers to the + # archives phase. + if repair: + return not (repo_only and (pack_errors or corrupt_ids)) + return not problems def list(self, limit=None, marker=None): """ diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 12abbd9aaf..b3cebf0292 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -594,6 +594,36 @@ def test_spoofed_manifest(archivers, request): cmd(archiver, "check", exit_code=0) +def test_check_repair_rebuilds_corrupt_index(archivers, request): + # A corrupt index with all packs intact: the default (full) --repair rebuilds the index from the + # packs and persists it (via the archives check, see ArchiveChecker.finish), leaving the repository + # usable again without a slow rebuild on the next access. + archiver = request.getfixturevalue(archivers) + check_cmd_setup(archiver) + cmd(archiver, "check", exit_code=0) + archive, repository = open_archive(archiver.repository_path, "archive1") + with repository: + assert isinstance(repository, Repository) + for info in repository.store_list("index"): # rot every index fragment + name = f"index/{info.name}" + data = bytearray(repository.store_load(name)) + data[0] ^= 0xFF + repository.store_store(name, bytes(data)) + cmd(archiver, "check", exit_code=1) # read-only check reports the corrupt index + output = cmd(archiver, "check", "-v", "--repair", exit_code=0) + assert "rebuilt" in output.lower() + # item 6: repair persisted a fresh index instead of leaving it for a slow rebuild on the next + # access. confirm the on-disk index exists and every fragment is intact. + archive, repository = open_archive(archiver.repository_path, "archive1") + with repository: + index_infos = list(repository.store_list("index")) + assert index_infos # a fresh index was persisted + for info in index_infos: # each fragment's content still matches its sha256 name + assert repository.store.hash(f"index/{info.name}") == info.name + cmd(archiver, "check", exit_code=0) # the repository is consistent again + assert "archive1" in cmd(archiver, "repo-list") # and remains usable + + @pytest.mark.skip(reason="TODO: repair does not yet rewrite store-corrupted packs, refs #8572") def test_manifest_rebuild_corrupted_chunk(archivers, request): archiver = request.getfixturevalue(archivers) diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 34d39646fb..18fe1c037e 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -1071,6 +1071,58 @@ def test_check_reports_invalid_pack_name(tmp_path, caplog): assert after.table[intact_id].result == 1 # the valid pack was checked +def test_check_repair_rebuilds_corrupt_index(tmp_path): + # check(repair=True) rebuilds a corrupt index from the packs' object headers. + location = os.fspath(tmp_path / "repo") + ids = [H(x) for x in range(10)] + with Repository(location, exclusive=True, create=True) as repository: + for i, cid in enumerate(ids): + repository.put(cid, fchunk(bytes([i]) * 20, chunk_id=cid)) + repository.flush() # seal the pack(s) and let close() persist the index + with reopen(repository) as repository: + index_names = [f"index/{info.name}" for info in repository.store_list("index")] + assert index_names # close() persisted at least one index fragment + for name in index_names: # rot every fragment so its content no longer matches its sha256 name + data = bytearray(repository.store_load(name)) + data[0] ^= 0xFF + repository.store_store(name, bytes(data)) + assert repository.check(repair=False) is False # read-only check reports the corrupt index + with reopen(repository) as repository: + assert repository.check(repair=True) is True # repair rebuilds the index from the packs + with reopen(repository) as repository: + assert repository.check(repair=False) is True # the rebuilt index passes a read-only check + for i, cid in enumerate(ids): + assert pdchunk(repository.get(cid)) == bytes([i]) * 20 # every chunk is indexed and resolves + + +def test_check_repair_refuses_when_pack_corrupt(tmp_path): + # A repair that finds any corrupt pack leaves the index and the pack untouched (no lossy rebuild, + # nothing dropped) and fails on a repository-only run, refs #8572, #10026. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(1), fchunk(b"GOOD-CHUNK", chunk_id=H(1))) + repository.flush() # seal a pack holding H(1) + repository.put(H(2), fchunk(b"LOST-CHUNK", chunk_id=H(2))) + repository.flush() # seal a separate pack holding H(2) + with reopen(repository) as repository: + bad_pack_name = "packs/" + bin_to_hex(repository.chunks[H(2)].pack_id) + data = bytearray(repository.store_load(bad_pack_name)) + data[-1] ^= 0xFF # rot the pack holding H(2): its content no longer matches its sha256 name + repository.store_store(bad_pack_name, bytes(data)) + for info in repository.store_list("index"): # rot the index so repair takes the rebuild path + name = f"index/{info.name}" + idata = bytearray(repository.store_load(name)) + idata[0] ^= 0xFF + repository.store_store(name, bytes(idata)) + with reopen(repository) as repository: + # a repository-only repair cannot fix a corrupt pack, so it fails. + assert repository.check(repair=True, repo_only=True) is False + # the corrupt pack is left in place, not dropped. + assert bad_pack_name in [f"packs/{info.name}" for info in repository.store_list("packs")] + with reopen(repository) as repository: + assert repository.check(repair=False) is False # index was not rebuilt; still corrupt + + def test_check_warns_on_invalid_chunk_index(tmp_path, caplog): # check warns about an invalid chunk index but does not fail, since the index is not part of # the repository's object integrity.