diff --git a/src/borg/archive.py b/src/borg/archive.py index 3299d07399..271118ac86 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -1849,6 +1849,11 @@ def __next__(self): class ArchiveChecker: + # Bound how many missing file chunks rebuild_archives buffers for its end-of-run report, + # so checking a badly damaged repo with very many missing chunks can not exhaust memory. + MAX_MISSING_CHUNKS = 1000 # max. distinct missing chunk ids kept for the grouped report + MAX_REFS_PER_CHUNK = 10 # max. referencing files kept per missing chunk + def __init__(self): self.error_found = False self.key = None @@ -2130,6 +2135,33 @@ def rebuild_archives( ): """Analyze and rebuild archives, expecting some damage and trying to make stuff consistent again.""" + # Missing file chunks, collected during the per-archive checks and reported grouped as + # chunk -> files -> archives after all archives were analyzed. Bounded by + # MAX_MISSING_CHUNKS / MAX_REFS_PER_CHUNK. + missing_chunks = {} # chunk_id -> (size, {path: {archive_name}}) + missing_chunks_truncated = False # True once the MAX_MISSING_CHUNKS cap was hit + missing_refs_truncated = set() # chunk_ids whose MAX_REFS_PER_CHUNK cap was hit + missing_refs_total = 0 # total missing chunk references seen (every file x chunk occurrence, uncapped) + + def record_missing_chunk(archive_name, path, chunk_id, size): + nonlocal missing_chunks_truncated, missing_refs_total + missing_refs_total += 1 + entry = missing_chunks.get(chunk_id) + if entry is None: + if len(missing_chunks) >= self.MAX_MISSING_CHUNKS: + missing_chunks_truncated = True + return + entry = missing_chunks[chunk_id] = (size, {}) + # one line per chunk id (not per file), so an interrupted check still logs what it found. + logger.error(f"Missing chunk detected: {bin_to_hex(chunk_id)}, {format_file_size(size)}.") + size, refs = entry + if path in refs: + refs[path].add(archive_name) + elif len(refs) < self.MAX_REFS_PER_CHUNK: + refs[path] = {archive_name} + else: + missing_refs_truncated.add(chunk_id) + def add_callback(chunk): id_ = self.key.id_hash(chunk) cdata = self.repo_objs.format(id_, {}, chunk, ro_type=ROBJ_ARCHIVE_STREAM) @@ -2146,16 +2178,17 @@ def add_reference(id_, size, cdata): self.chunks.update_pack_info(pack_results) def verify_file_chunks(archive_name, item): - """Verifies that all file chunks are present. Missing file chunks will be logged.""" + """Verify that all of a file's chunks are present, collecting any missing ones for the report.""" offset = 0 for chunk in item.chunks: chunk_id, size = chunk if chunk_id not in self.chunks: - logger.error( + logger.debug( "{}: {}: Missing file chunk detected (Byte {}-{}, Chunk {}).".format( archive_name, item.path, offset, offset + size, bin_to_hex(chunk_id) ) ) + record_missing_chunk(archive_name, item.path, chunk_id, size) self.error_found = True offset += size if "size" in item: @@ -2169,6 +2202,24 @@ def verify_file_chunks(archive_name, item): ) ) + def report_missing_chunks(): + """Report the collected missing chunks, grouped as chunk -> files -> archives.""" + if not missing_chunks: + return + logger.error("The following chunks are missing in the repository:") + for chunk_id, (size, refs) in missing_chunks.items(): + logger.error(f"- Chunk {bin_to_hex(chunk_id)}, {format_file_size(size)}") + for path in sorted(refs): + archive_names = ", ".join(sorted(refs[path])) + logger.error(f" - {path}: {archive_names}") + if chunk_id in missing_refs_truncated: + logger.error(f" - ... (only the first {self.MAX_REFS_PER_CHUNK} files are listed)") + if missing_chunks_truncated: + logger.error( + f"... (only the first {self.MAX_MISSING_CHUNKS} missing chunks are listed; " + f"{missing_refs_total} missing chunk references total)" + ) + def robust_iterator(archive): """Iterates through all archive items @@ -2270,62 +2321,68 @@ def valid_item(obj): pi = ProgressIndicatorPercent( total=num_archives, msg="Checking archives %3.1f%%", step=0.1, msgid="check.rebuild_archives" ) - for i, info in enumerate(archive_infos): - pi.show(i) - archive_id, archive_id_hex = info.id, bin_to_hex(info.id) - try: - formatted = formatter.format_item(info, jsonline=False) - except (Archive.DoesNotExist, Repository.ObjectNotFound, IntegrityErrorBase): - # keys like {comment} need the archive metadata, which is damaged or missing here. - # use the values from the archive directory entry, they are always available. - formatted = f"{info.name} {OutputTimestamp(info.ts)} {archive_id_hex}" - logger.info(f"Analyzing archive {formatted} ({i + 1}/{num_archives})") - if archive_id not in self.chunks: - logger.error(f"Archive metadata block {archive_id_hex} is missing!") - self.error_found = True - if self.repair: - logger.error(f"Deleting broken archive {info.name} {archive_id_hex}.") - self.manifest.archives.delete_by_id(archive_id) - else: - logger.error(f"Would delete broken archive {info.name} {archive_id_hex}.") - continue - cdata = self.repository.get(archive_id) - try: - _, data = self.repo_objs.parse(archive_id, cdata, ro_type=ROBJ_ARCHIVE_META) - except IntegrityErrorBase as integrity_error: - logger.error(f"Archive metadata block {archive_id_hex} is corrupted: {integrity_error}") - self.error_found = True + # report the missing chunks collected so far even if the loop is interrupted (Ctrl-C) or aborts + # with an exception (e.g. the "Unknown archive metadata version" raise below), so a check of a + # badly damaged repo does not throw away everything it already found. + try: + for i, info in enumerate(archive_infos): + pi.show(i) + archive_id, archive_id_hex = info.id, bin_to_hex(info.id) + try: + formatted = formatter.format_item(info, jsonline=False) + except (Archive.DoesNotExist, Repository.ObjectNotFound, IntegrityErrorBase): + # keys like {comment} need the archive metadata, which is damaged or missing here. + # use the values from the archive directory entry, they are always available. + formatted = f"{info.name} {OutputTimestamp(info.ts)} {archive_id_hex}" + logger.info(f"Analyzing archive {formatted} ({i + 1}/{num_archives})") + if archive_id not in self.chunks: + logger.error(f"Archive metadata block {archive_id_hex} is missing!") + self.error_found = True + if self.repair: + logger.error(f"Deleting broken archive {info.name} {archive_id_hex}.") + self.manifest.archives.delete_by_id(archive_id) + else: + logger.error(f"Would delete broken archive {info.name} {archive_id_hex}.") + continue + cdata = self.repository.get(archive_id) + try: + _, data = self.repo_objs.parse(archive_id, cdata, ro_type=ROBJ_ARCHIVE_META) + except IntegrityErrorBase as integrity_error: + logger.error(f"Archive metadata block {archive_id_hex} is corrupted: {integrity_error}") + self.error_found = True + if self.repair: + logger.error(f"Deleting broken archive {info.name} {archive_id_hex}.") + self.manifest.archives.delete_by_id(archive_id) + else: + logger.error(f"Would delete broken archive {info.name} {archive_id_hex}.") + continue + archive = self.key.unpack_archive(data) + archive = ArchiveItem(internal_dict=archive) + if archive.version != 2: + raise Exception("Unknown archive metadata version") + items_buffer = ChunkBuffer(self.key) + items_buffer.write_chunk = add_callback + for item in robust_iterator(archive): + if "chunks" in item: + verify_file_chunks(info.name, item) + items_buffer.add(item) + items_buffer.flush(flush=True) if self.repair: - logger.error(f"Deleting broken archive {info.name} {archive_id_hex}.") - self.manifest.archives.delete_by_id(archive_id) - else: - logger.error(f"Would delete broken archive {info.name} {archive_id_hex}.") - continue - archive = self.key.unpack_archive(data) - archive = ArchiveItem(internal_dict=archive) - if archive.version != 2: - raise Exception("Unknown archive metadata version") - items_buffer = ChunkBuffer(self.key) - items_buffer.write_chunk = add_callback - for item in robust_iterator(archive): - if "chunks" in item: - verify_file_chunks(info.name, item) - items_buffer.add(item) - items_buffer.flush(flush=True) - if self.repair: - archive.item_ptrs = archive_put_items( - items_buffer.chunks, repo_objs=self.repo_objs, add_reference=add_reference - ) - data = self.key.pack_metadata(archive.as_dict()) - new_archive_id = self.key.id_hash(data) - logger.debug(f"archive id old: {bin_to_hex(archive_id)}") - logger.debug(f"archive id new: {bin_to_hex(new_archive_id)}") - cdata = self.repo_objs.format(new_archive_id, {}, data, ro_type=ROBJ_ARCHIVE_META) - add_reference(new_archive_id, len(data), cdata) - self.manifest.archives.create(info.name, new_archive_id, info.ts) - if archive_id != new_archive_id: - self.manifest.archives.delete_by_id(archive_id) - pi.finish() + archive.item_ptrs = archive_put_items( + items_buffer.chunks, repo_objs=self.repo_objs, add_reference=add_reference + ) + data = self.key.pack_metadata(archive.as_dict()) + new_archive_id = self.key.id_hash(data) + logger.debug(f"archive id old: {bin_to_hex(archive_id)}") + logger.debug(f"archive id new: {bin_to_hex(new_archive_id)}") + cdata = self.repo_objs.format(new_archive_id, {}, data, ro_type=ROBJ_ARCHIVE_META) + add_reference(new_archive_id, len(data), cdata) + self.manifest.archives.create(info.name, new_archive_id, info.ts) + if archive_id != new_archive_id: + self.manifest.archives.delete_by_id(archive_id) + finally: + pi.finish() + report_missing_chunks() def finish(self): if self.repair: diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index ef42ad776b..64c3fc990f 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -6,7 +6,7 @@ import pytest -from ...archive import ChunkBuffer +from ...archive import ChunkBuffer, ArchiveChecker from ...constants import * # NOQA from ...helpers import bin_to_hex, msgpack, CommandError, IntegrityError from ...manifest import Manifest @@ -16,6 +16,7 @@ cmd, src_file, create_src_archive, + create_regular_file, open_archive, generate_archiver_tests, read_chunk, @@ -178,9 +179,20 @@ def test_missing_file_chunk(archivers, request): pytest.fail("should not happen") # convert 'fail' output = cmd(archiver, "check", exit_code=1) - assert "Missing file chunk detected" in output + assert "The following chunks are missing in the repository:" in output + # archive1 and archive2 share src_file, so the missing chunk is grouped once, with both archives + # listed on its single reference line (the id also appears once in the streamed "Missing chunk + # detected" line emitted while the archives are analyzed). + killed_hex = bin_to_hex(killed_chunk.id) + chunk_header_lines = [ln for ln in output.splitlines() if ln.startswith("- Chunk ") and killed_hex in ln] + assert len(chunk_header_lines) == 1 + ref_lines = [line for line in output.splitlines() if src_file in line] + assert len(ref_lines) == 1 + assert "archive1" in ref_lines[0] and "archive2" in ref_lines[0] output = cmd(archiver, "check", "--repair", exit_code=0) - assert "Missing file chunk detected" in output # repair is not changing anything, just reporting. + # repair is not changing anything, just reporting. + assert "The following chunks are missing in the repository:" in output + assert bin_to_hex(killed_chunk.id) in output # check does not modify the chunks list. for archive_name in ("archive1", "archive2"): @@ -200,7 +212,63 @@ def test_missing_file_chunk(archivers, request): # check should not complain anymore about missing chunks: output = cmd(archiver, "check", "-v", "--repair", exit_code=0) - assert "Missing file chunk detected" not in output + assert "The following chunks are missing in the repository:" not in output + + +def test_missing_file_chunk_report_truncated(archiver): + # local-only: this patches ArchiveChecker.MAX_MISSING_CHUNKS in-process, which has no effect + # when borg runs as a separate process (binary_archiver), so it must not be parametrized. + check_cmd_setup(archiver) + + # remove several distinct file chunks, so more missing chunks exist than the (patched) report limit. + archive, repository = open_archive(archiver.repository_path, "archive1") + killed_ids = [] + with repository: + for item in archive.iter_items(): + if "chunks" not in item or not item.chunks: + continue + chunk_id = item.chunks[-1].id + if chunk_id not in killed_ids: + repository.delete(chunk_id) + killed_ids.append(chunk_id) + if len(killed_ids) >= 3: + break + assert len(killed_ids) >= 2 # need several distinct missing chunks to exercise truncation + + # cap the report to a single chunk, so the remaining missing chunks are truncated. + with patch.object(ArchiveChecker, "MAX_MISSING_CHUNKS", 1): + output = cmd(archiver, "check", exit_code=1) + assert "The following chunks are missing in the repository:" in output + assert output.count("- Chunk ") == 1 # only one chunk is detailed + assert "only the first 1 missing chunks are listed" in output # the rest are noted as truncated + + +def test_missing_file_chunk_refs_truncated(archivers, request): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + + # many distinct files with identical content dedup to the same chunk, so a single missing chunk + # ends up referenced by more files than MAX_REFS_PER_CHUNK, which exercises the per-chunk cap + # without patching (so it works in binary mode too, where borg runs as a separate process). + cap = ArchiveChecker.MAX_REFS_PER_CHUNK + for i in range(cap + 1): + create_regular_file(archiver.input_path, f"samefile{i}", contents=b"same content for dedup") + cmd(archiver, "create", "archive1", "input") + + archive, repository = open_archive(archiver.repository_path, "archive1") + killed_id = None + with repository: + for item in archive.iter_items(): + if item.path.endswith("samefile0"): + killed_id = item.chunks[0].id + repository.delete(killed_id) + break + assert killed_id is not None + + output = cmd(archiver, "check", exit_code=1) + assert "The following chunks are missing in the repository:" in output + assert bin_to_hex(killed_id) in output + assert f"only the first {cap} files are listed" in output # the remaining referencing files are truncated def test_missing_archive_item_chunk(archivers, request): @@ -483,11 +551,14 @@ def test_verify_data(archivers, request, init_args): # repair will find the defect chunk and remove it output = cmd(archiver, "check", "--repair", "--verify-data", exit_code=0) assert f"{bin_to_hex(chunk.id)}, integrity error" in output - assert f"{src_file}: Missing file chunk detected" in output + assert "The following chunks are missing in the repository:" in output + assert bin_to_hex(chunk.id) in output + assert src_file in output # run with --verify-data again, it will notice the missing chunk. output = cmd(archiver, "check", "--archives-only", "--verify-data", exit_code=1) - assert f"{src_file}: Missing file chunk detected" in output + assert "The following chunks are missing in the repository:" in output + assert bin_to_hex(chunk.id) in output def test_verify_data_wrong_chunk_content(archivers, request, monkeypatch): @@ -571,12 +642,15 @@ def test_corrupted_file_chunk(archivers, request, init_args): # repair: the defect chunk will be removed. output = cmd(archiver, "check", "--repair", "--verify-data", exit_code=0) assert f"{bin_to_hex(chunk.id)}, integrity error" in output - assert f"{src_file}: Missing file chunk detected" in output + assert "The following chunks are missing in the repository:" in output + assert bin_to_hex(chunk.id) in output + assert src_file in output # run normal check again cmd(archiver, "check", "--repository-only", exit_code=0) output = cmd(archiver, "check", "--archives-only", exit_code=1) - assert f"{src_file}: Missing file chunk detected" in output + assert "The following chunks are missing in the repository:" in output + assert src_file in output @pytest.mark.skip(