From f8f14648f0f41acbbfd2fcd53bf86a2de61cd734 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Wed, 16 Sep 2026 17:56:52 +0530 Subject: [PATCH 1/2] remove the drop_corrupt_tail mechanism, refs #10369 Every check --repair rebuild of the chunk index has an object validator (#10369), so drop_corrupt_tail was only reachable from tests. - PackReader.iter_headers: remove the drop_corrupt_tail parameter. Without a validator, a corrupt object header raises IntegrityError. - build_chunkindex_from_repo: remove the drop_corrupt_tail parameter. - Repository: remove chunkindex_drop_corrupt_tail and chunkindex_validate. Since #10368 the checker hands its index to the repository, so the lazy .chunks rebuild never runs during a check and nothing set either of them. - ArchiveChecker.check: stop passing drop_corrupt_tail. - tests: remove the 5 tests for drop_corrupt_tail, rewrite test_check_without_repair_does_not_drop_a_pack_tail as test_check_without_key_aborts_on_a_corrupt_pack_header. --- src/borg/archive.py | 7 +--- src/borg/cache.py | 17 +++------ src/borg/repository.py | 38 +++++-------------- src/borg/testsuite/archiver/check_cmd_test.py | 18 ++++----- src/borg/testsuite/cache_test.py | 34 ----------------- src/borg/testsuite/repository_test.py | 37 ------------------ 6 files changed, 24 insertions(+), 127 deletions(-) diff --git a/src/borg/archive.py b/src/borg/archive.py index 64ab905b82..195fcb86d0 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -2285,14 +2285,11 @@ def check( self.chunks = build_chunkindex_from_repo( self.repository, slow_rebuild=repair, + # validate is None only without --repair and without the key: a corrupt object header then + # raises CorruptPack. validate=validate, # dropped content is a check finding, with or without --repair. on_drop=self.note_dropped_objects, - # without a validator the rebuild can not resync past a corrupt object header. --repair - # drops the rest of that pack to get on with the repair; without --repair the rebuild - # raises, so an index missing objects that are still there can not make the check report - # them as gone. - drop_corrupt_tail=repair, write_immediately=False, ) # clear F_NEW (entry not in the index/ fragments yet), so Repository.close() does not store diff --git a/src/borg/cache.py b/src/borg/cache.py index 30f91dbd3d..92933dc001 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -884,20 +884,15 @@ def build_chunkindex_from_repo( fragments_only=False, validate=None, on_drop=None, - drop_corrupt_tail=False, write_immediately=False, init_flags=ChunkIndex.F_USED, ): # fragments_only: build the index from the index/ fragments only, returning None if they cannot be # read completely, and never write to the repo. - # validate: a repo object validator, handed to PackReader.iter_headers so the rebuild skips the - # objects that fail it. - # on_drop: a callable, handed to PackReader.iter_headers, which calls it once per place where - # 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. + # validate: a repo object validator or None, passed to PackReader.iter_headers. With a validator, + # the rebuild skips the objects that fail it; without one, a corrupt object header raises CorruptPack. + # on_drop: a callable or None, passed to PackReader.iter_headers, called once per byte range the + # validating walk skips. 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: @@ -994,9 +989,7 @@ def build_chunkindex_from_repo( pack_id = hex_to_bin(info.name) reader = PackReader(repository.store, pack_id) try: - for chunk_id, obj_offset, obj_size in reader.iter_headers( - validate=validate, on_drop=on_drop, drop_corrupt_tail=drop_corrupt_tail - ): + for chunk_id, obj_offset, obj_size in reader.iter_headers(validate=validate, on_drop=on_drop): num_chunks += 1 chunks[chunk_id] = ChunkIndexEntry( flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size diff --git a/src/borg/repository.py b/src/borg/repository.py index 3ff393bc42..5435f1d040 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -494,7 +494,7 @@ def _find_header(self, offset, pack_size, validate): offset += max(len(buf) - (hdr_size - 1), 1) return None - def iter_headers(self, validate=None, on_drop=None, drop_corrupt_tail=False): + def iter_headers(self, validate=None, on_drop=None): """Yield (chunk_id, offset, size) for each object by walking the fixed object headers. The walk reads one range per object (or a slice, for a pack in memory), plus one store @@ -510,13 +510,11 @@ def iter_headers(self, validate=None, on_drop=None, drop_corrupt_tail=False): validates, the walk yields nothing and raises nothing. Without a validator a resync is impossible, because payload bytes can look like a header. - A header that _parse_header rejects then raises IntegrityError naming what is wrong with - it, or, with drop_corrupt_tail, ends the walk there and drops the rest of the pack. + A header that _parse_header rejects then raises IntegrityError naming what is wrong with it. - on_drop, if given, is called once per place where the walk discards content: once for the - object with the failed header plus whatever the resync scan skips before the object it - resumes at, once for a tail dropped because the scan found no such object or because there - was no validator to scan with. It only reports, it does not change what the walk does. + on_drop, if given, is called once per byte range a validating walk skips: the object with + the failed header plus the bytes up to the next object validate accepts, or up to the end of + the pack if there is none. headers_parsed is set to the number of headers _parse_header accepted in this walk, the candidates the resync scan tried included. A pack whose bytes hold no object header at all @@ -542,19 +540,9 @@ def iter_headers(self, validate=None, on_drop=None, drop_corrupt_tail=False): problem = self._validation_problem(hdr, offset, buf, offset, validate) if problem is not None: if validate is None: - # 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: - # 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( - f"pack {pack_hex}: {problem} at offset {offset}, no validator to resync with, " - f"skipping the remaining {pack_size - offset} bytes." - ) - break + # without a validator, payload bytes that look like a header can not be told + # apart from an object, so the walk can not continue past this header. + raise IntegrityError(f"pack {pack_hex}: {problem} at offset {offset} (pack corruption)") if on_drop is not None: on_drop() # content is discarded either way below: this object, or the tail. found = self._find_header(offset + 1, pack_size, validate) @@ -976,12 +964,6 @@ 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 (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. TODO(#10378): nothing sets them, remove both. - 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) @@ -1277,9 +1259,7 @@ def chunks(self): if self._chunks is None: from .cache import build_chunkindex_from_repo - self._chunks = build_chunkindex_from_repo( - self, validate=self.chunkindex_validate, drop_corrupt_tail=self.chunkindex_drop_corrupt_tail - ) + self._chunks = build_chunkindex_from_repo(self) return self._chunks @chunks.setter diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index bdfc14dbb4..5deab28b9f 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -964,12 +964,11 @@ def test_check_repair_validates_index_rebuild(archivers, request): assert neighbour_id in repository.chunks -def test_check_without_repair_does_not_drop_a_pack_tail(archivers, request, monkeypatch): - """A check without --repair reports a corrupt object header, it does not index the pack up to it. +def test_check_without_key_aborts_on_a_corrupt_pack_header(archivers, request, monkeypatch): + """A check without --repair and without the key raises CorruptPack at a corrupt object header. - Without a validator the walk can not resync past a corrupt object header. --repair passes - drop_corrupt_tail, so the rest of that pack is dropped and the repair gets on; a check without - --repair passes drop_corrupt_tail=False and the walk raises instead. + Without the key there is no object validator, and without one the pack walk raises at a corrupt + object header. The rebuild only walks the packs when the chunk index fragments are unusable, and it only walks without a validator when the key can not be read, so the test arranges both. @@ -1013,9 +1012,9 @@ def build_chunkindex_from_repo(repository, **kwargs): try: index = real_build(repository, **kwargs) except Exception as err: - rebuilds.append((kwargs.get("drop_corrupt_tail"), err)) + rebuilds.append((kwargs.get("validate"), err)) raise - rebuilds.append((kwargs.get("drop_corrupt_tail"), index)) + rebuilds.append((kwargs.get("validate"), index)) return index monkeypatch.setattr(ArchiveChecker, "make_key", make_key) @@ -1025,10 +1024,9 @@ def build_chunkindex_from_repo(repository, **kwargs): 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 + validate, outcome = rebuilds[0] + assert validate is None 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): diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index a92cd6559f..e4425fef1a 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -546,40 +546,6 @@ def test_build_chunkindex_reports_a_pack_without_any_object_header(tmp_path): assert len(drops) == 1 -def test_build_chunkindex_without_a_validator_drops_the_rest_of_a_damaged_pack(tmp_path): - """With drop_corrupt_tail and no validator, a corrupt object header ends the pack's walk.""" - from .repository_test import fchunk - - obj1 = fchunk(b"first", chunk_id=H(90)) - obj2 = bytearray(fchunk(b"second", chunk_id=H(91))) - obj2[0] ^= 0xFF # break the magic of the second object's header - obj3 = fchunk(b"third", chunk_id=H(92)) - 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) + obj3) - index = build_chunkindex_from_repo( - repository, slow_rebuild=True, on_drop=lambda: drops.append(True), drop_corrupt_tail=True - ) - assert H(90) in index # the pack is indexed up to the damaged header - assert H(91) not in index and H(92) not in index # from there on the pack is dropped - assert len(drops) == 1 - - -def test_build_chunkindex_without_drop_corrupt_tail_raises_on_a_damaged_pack(tmp_path): - """on_drop alone does not let the rebuild past a corrupt object header, it only reports.""" - from .repository_test import fchunk - - obj1 = fchunk(b"first", chunk_id=H(90)) - obj2 = bytearray(fchunk(b"second", chunk_id=H(91))) - obj2[0] ^= 0xFF # break the magic of the second object's header - 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(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 - - def test_build_chunkindex_drops_a_pack_that_validates_nothing_when_others_do(tmp_path): """A single pack of which nothing validates is dropped, the objects of the other packs are indexed.""" from .repository_test import fchunk diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 80be75677f..b56b957455 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -2206,43 +2206,6 @@ def test_pack_reader_raises_on_bad_magic(): list(reader.iter_headers()) -def test_pack_reader_drops_a_corrupt_tail_only_when_asked(): - # without a validator there is nothing to resync with, so the walk can not get past a corrupt - # header. drop_corrupt_tail alone decides what happens then; on_drop only reports it. - obj1 = fchunk(b"payload-one", chunk_id=H(1)) - obj2 = bytearray(fchunk(b"payload-two", chunk_id=H(2))) - obj2[0] ^= 0xFF # break the magic of the second object's header - pack = obj1 + bytes(obj2) - drops = [] - reader = PackReader(pack_contents=pack) - with pytest.raises(IntegrityError, match="no object header at offset"): - list(reader.iter_headers(on_drop=lambda: drops.append(1))) - assert drops == [] # nothing was discarded: the walk raised instead - headers = list(reader.iter_headers(on_drop=lambda: drops.append(1), drop_corrupt_tail=True)) - assert headers == [(H(1), 0, len(obj1))] # up to the corrupt header, the rest of the pack is gone - assert len(drops) == 1 - - -def test_pack_reader_drops_a_corrupt_tail_without_an_on_drop(): - # drop_corrupt_tail works without an on_drop to report the drop to. - obj1 = fchunk(b"payload-one", chunk_id=H(1)) - obj2 = bytearray(fchunk(b"payload-two", chunk_id=H(2))) - obj2[0] ^= 0xFF - reader = PackReader(pack_contents=obj1 + bytes(obj2)) - assert list(reader.iter_headers(drop_corrupt_tail=True)) == [(H(1), 0, len(obj1))] - - -def test_pack_reader_drop_corrupt_tail_does_not_affect_a_validating_walk(): - # with a validator the walk resyncs, so drop_corrupt_tail changes nothing. - obj1 = bytearray(fchunk(b"payload-one", chunk_id=H(1))) - obj2 = fchunk(b"payload-two", chunk_id=H(2)) - obj1[0] ^= 0xFF - reader = PackReader(pack_contents=bytes(obj1) + obj2) - resynced = [(H(2), len(obj1), len(obj2))] - assert list(reader.iter_headers(validate=accept_all)) == resynced - assert list(reader.iter_headers(validate=accept_all, drop_corrupt_tail=True)) == resynced - - def test_pack_reader_raises_on_bad_magic_through_store(tmp_path): obj = bytearray(fchunk(b"FIRST", chunk_id=H(47))) obj[0] ^= 0xFF From 8903d7deb5cbecf1432713c316e16f0cf0ebbc24 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Wed, 16 Sep 2026 16:18:42 +0530 Subject: [PATCH 2/2] check --repair: re-read only the packs the repair wrote, refs #8466 finish() validates the written packs against the shared index instead of rebuilding it from all packs. --- src/borg/archive.py | 151 ++++++-- src/borg/repository.py | 27 +- src/borg/testsuite/archiver/check_cmd_test.py | 326 +++++++++++++++++- src/borg/testsuite/repository_test.py | 12 + 4 files changed, 461 insertions(+), 55 deletions(-) diff --git a/src/borg/archive.py b/src/borg/archive.py index 195fcb86d0..cfbc02352f 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -53,7 +53,8 @@ from .patterns import PathPrefixPattern, FnmatchPattern, IECommand from .item import Item, ArchiveItem, ItemDiff from .platform import acl_get, acl_set, set_flags, get_flags, set_times, swidth -from .repository import Repository +from .hashindex import ChunkIndex, ChunkIndexEntry +from .repository import Repository, PackReader from .repoobj import RepoObj, object_validator # macOS: SF_DATALESS marks dataless placeholder files (e.g. cloud files not materialized locally). @@ -2212,9 +2213,24 @@ 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. + # True once repair wrote a pack: it stored a chunk or deleted a defect chunk. self.chunks_modified = False + # ids of the existing packs repair stored (put(), flush()) or wrote by rewriting a pack (delete()). + self.written_packs = set() + + def record_stored(self, results): + """Add the pack ids in results to written_packs. + + results: the (chunk_id, pack_id, obj_offset, obj_size) tuples Repository.put() or .flush() returns + for the packs it stored, or None if it stored no pack. + """ + if results: + self.written_packs.update(pack_id for _, pack_id, _, _ in results) + + def create_archive_entry(self, name, id, ts): + """Store the pack writer buffer, record the packs it wrote, create the archives directory entry.""" + self.record_stored(self.repository.flush()) + self.manifest.archives.create(name, id, ts) def note_dropped_objects(self): # The chunk index rebuild skipped repository content to get past a corrupt object header. @@ -2273,6 +2289,7 @@ def check( validate = object_validator(self.repo_objs) else: validate = None + assert not repair or validate is not None # a repair validates every object it indexes # store the chunks buffered in the pack writer, so the index below has their pack locations # (pack id, offset and size in the pack). self.repository.flush() @@ -2405,9 +2422,14 @@ def verify_data(self): # failed twice -> remove this defect chunk. delete rewrites its pack without it, # keeping the other chunks, and removes it from self.chunks, so rebuild_archives # reports the file it belongs to. update_index=False: finish() stores the index - # rebuilt from the packs and clears the invalid marker delete() writes. - self.repository.delete(defect_chunk, update_index=False, validate=validate) + # and clears the invalid marker delete() writes. + # new_pack_id holds the other objects of the old pack, None if there were none. + old_pack_id = self.chunks[defect_chunk].pack_id + new_pack_id, _ = self.repository.delete(defect_chunk, update_index=False, validate=validate) self.chunks_modified = True + self.written_packs.discard(old_pack_id) + if new_pack_id is not None: + self.written_packs.add(new_pack_id) else: logger.warning("chunk %s not deleted, did not consistently fail.", bin_to_hex(defect_chunk)) else: @@ -2495,7 +2517,7 @@ def valid_archive(obj): self.error_found = True if self.repair: logger.warning(f"Creating archives directory entry for {name} {archive_id_hex}.") - self.manifest.archives.create(name, archive_id, archive.time) + self.create_archive_entry(name, archive_id, archive.time) else: logger.warning(f"Would create archives directory entry for {name} {archive_id_hex}.") @@ -2551,7 +2573,7 @@ def add_reference(id_, size, cdata): # with --repair, store a chunk the repository does not have; put() adds it to self.chunks. if self.repair and id_ not in self.chunks: assert cdata is not None - self.repository.put(id_, cdata) + self.record_stored(self.repository.put(id_, cdata)) self.chunks_modified = True def verify_file_chunks(archive_name, item): @@ -2761,43 +2783,108 @@ def valid_item(obj): 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) + self.create_archive_entry(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 verify_written_packs(self): + """Read the object headers of the packs in written_packs and make the chunks index match them. + + put() and delete() compute the index entries of the packs they write without reading the packs. + This compares the (chunk_id, obj_offset, obj_size) of each object header in a written pack, read + with a validator, with the index entries that name the pack. Each difference is a check finding, + logged and fixed in the index: + + - an index entry names an object the pack does not hold: the entry is removed. + - the pack holds an object whose chunk id is not indexed: the object is indexed. + - the pack does not exist: its index entries are removed. + + An object whose chunk id is indexed at another location is a superseded duplicate, not a finding: a + pack delete() wrote can hold one, in a byte range compact_pack copied with no index entry covering it. + """ + pack_ids = sorted(self.written_packs) + if not pack_ids: + return + logger.info(f"Re-reading the packs written by the repair: {len(pack_ids)}.") + # (chunk_id, obj_offset, obj_size) of the index entries, per written pack. + indexed = {pack_id: set() for pack_id in pack_ids} + for chunk_id, entry in self.chunks.iteritems(): + entries = indexed.get(entry.pack_id) + if entries is not None: + entries.add((chunk_id, entry.obj_offset, entry.obj_size)) + validate = object_validator(self.repo_objs) + for pack_id in pack_ids: + # PackReader reads from the store, which does not refresh the repository lock. + self.repository._lock_refresh() + pack_hex = bin_to_hex(pack_id) + expected = indexed.pop(pack_id) + reader = PackReader(self.repository.store, pack_id) + # iter_headers() yields nothing for a missing pack: the store reports size 0 for it. + if not self.repository.store.info(reader.key).exists: + self.error_found = True + logger.error(f"pack {pack_hex}: written by the repair, but it is missing. Removing its index entries.") + for chunk_id, _, _ in expected: + del self.chunks[chunk_id] + continue + found = list(reader.iter_headers(validate=validate, on_drop=self.note_dropped_objects)) + not_found = sorted(expected.difference(found)) + for chunk_id, _, _ in not_found: + del self.chunks[chunk_id] + # the loop indexes each unindexed object, so of several unindexed copies of a chunk, the first + # is indexed and the others are superseded duplicates. + unindexed = [] + for obj in found: + chunk_id, obj_offset, obj_size = obj + if obj in expected: + continue + if chunk_id in self.chunks: + logger.debug( + f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: " + "superseded duplicate" + ) + continue + unindexed.append(obj) + # size=0: the object header does not hold the plaintext size. + self.chunks[chunk_id] = ChunkIndexEntry( + flags=ChunkIndex.F_USED, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size + ) + if not (not_found or unindexed): + continue + self.error_found = True + logger.error( + f"pack {pack_hex}: the chunks index does not match the pack. Indexed objects not in the pack: " + f"{len(not_found)}, objects in the pack with an unindexed chunk id: {len(unindexed)}. " + "Fixed the index." + ) + for chunk_id, obj_offset, obj_size in not_found: + logger.debug( + f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: not in pack" + ) + for chunk_id, obj_offset, obj_size in unindexed: + logger.debug( + f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: not indexed" + ) + def finish(self): if self.repair: - # flush chunks re-added during repair so their packs are on the store and out of the pack - # writer buffer (close() requires an empty buffer, #10055) before we (re)build the index. - self.repository.flush() + # store the pack writer buffer before the index is written (close() requires an empty buffer, #10055). + self.record_stored(self.repository.flush()) if self.chunks_modified: - # the packs changed: rebuild the index from them and store it. The index/ fragments lack - # the chunks this repair stored, so the index is invalid until the rebuilt one is stored. - # Free the current index first, so only one index is in memory. + # the index/ fragments do not have the chunks this repair stored. write_chunkindex_invalid(self.repository) - self.repository.invalidate_chunk_index() - self.chunks = None - logger.info("Rebuilding and writing the repository chunks index.") - build_chunkindex_from_repo( - self.repository, - slow_rebuild=True, - validate=object_validator(self.repo_objs), - on_drop=self.note_dropped_objects, - 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 - ) + self.verify_written_packs() + 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 + ) + # close() persists the in-memory index: drop it, the stored one is current. + self.repository.invalidate_chunk_index() + self.chunks = None # the stored index matches the packs: clear the invalid marker. delete_chunkindex_invalid(self.repository) - # drop the in-memory index so close() does not persist it over the index just written. - self.repository.invalidate_chunk_index() class ArchiveRecreater: diff --git a/src/borg/repository.py b/src/borg/repository.py index 5435f1d040..ae5995b012 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -1290,10 +1290,15 @@ def is_chunk_index_loaded(self): return self._chunks is not None def flush(self): - """Flush any buffered pack writer chunks.""" + """Store the pack writer buffer as a pack, after waiting for the pack the background store-thread is storing. + + Returns the (chunk_id, pack_id, obj_offset, obj_size) tuples of the objects in the packs this call + stored or waited for, or None if there were none. + """ if self._pack_writer is not None: self._lock_refresh() - self._pack_writer.flush() # PackWriter updates _chunks internally + return self._pack_writer.flush() # PackWriter updates _chunks internally + return None def close(self, *, aborting=False): """Close the repository: join an in-flight pack store, persist the chunk index, tear down. @@ -1376,12 +1381,10 @@ def check(self, repair=False, max_duration=0, max_age=0, repo_only=False, valida 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 the store hash, which is content-addressing rather - than a MAC, so that check detects accidental corruption but not tampering; the rebuild therefore - checks every object with validate, see below, 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. + persisted. Packs are verified by the store hash, which is content-addressing rather than a MAC, so + that check detects accidental corruption but not tampering; the rebuild therefore checks every + object with validate, see below, 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 @@ -1809,9 +1812,12 @@ def delete(self, id, *, validate, update_index=True): Raises PermissionDenied before any store change unless the repo permissions grant write and delete on packs/ and index/ (see assert_writable). + validate: passed to compact_pack. update_index: True: store the full chunk index and delete the invalid marker. False: update the in-memory index only; the marker stays until the index is stored and the marker deleted. - validate: passed to compact_pack. + + Returns compact_pack's (new_pack_id, dropped_bytes): the id of the pack holding the other objects + of the old pack (None if there were none), and the number of bytes the rewrite dropped. """ from .cache import write_chunkindex_to_repo, write_chunkindex_invalid, delete_chunkindex_invalid @@ -1824,7 +1830,7 @@ def delete(self, id, *, validate, update_index=True): # keep every object the chunk index lists for this pack, except the one being deleted. keep_ids = {cid for cid, e in self.chunks.iteritems() if e.pack_id == pack_id} keep_ids.discard(id) - self.compact_pack( + result = self.compact_pack( pack_id, keep_ids=keep_ids, drop_ids={id}, @@ -1836,6 +1842,7 @@ def delete(self, id, *, validate, update_index=True): # the removal for the next borg process. write_chunkindex_to_repo(self, self.chunks, incremental=False, force_write=True, delete_other=True) delete_chunkindex_invalid(self) + return result def compact_pack( self, pack_id, *, keep_ids: set, drop_ids: set, validate, chunks=None, before_old_pack_delete=None diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 5deab28b9f..6e8f47d628 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -28,7 +28,7 @@ from ...item import Item from ...manifest import Archives, Manifest from ...repoobj import RepoObj -from ...repository import PackTracker, Repository +from ...repository import PackReader, PackTracker, Repository from ..repoobj_test import DATA_SIZE_OFFSET from ..repository_test import fchunk, corrupt_chunk_on_disk from . import ( @@ -483,10 +483,11 @@ def test_missing_archive_metadata(archivers, request): # checker_builds: per index build in ArchiveChecker, whether repository.chunks was loaded at that time. -# A full check without --repair uses the index the repository check loaded, --repair also builds in finish(). +# A full check without --repair uses the index the repository check loaded. --repair builds once: finish() +# re-reads only the packs the repair wrote, see test_repair_finish_reads_only_the_packs_put_wrote. @pytest.mark.parametrize( "args, exit_code, checker_builds", - [(["--archives-only"], 1, [False]), ([], 1, []), (["--repair"], 0, [False, False])], + [(["--archives-only"], 1, [False]), ([], 1, []), (["--repair"], 0, [False])], ids=["archives-only", "full", "repair"], ) def test_check_holds_a_single_chunk_index(archiver, monkeypatch, args, exit_code, checker_builds): @@ -625,24 +626,21 @@ def rebuild_archives(self, **kwargs): repository.get(chunk_id) -def test_check_repair_stopped_in_the_index_rebuild_marks_the_index_invalid(archiver, monkeypatch): - """A --repair check that stops in the index rebuild of finish() leaves the chunk index marked invalid. +def test_check_repair_stopped_in_the_index_store_marks_the_index_invalid(archiver, monkeypatch): + """A --repair check that stops while finish() stores the chunk index leaves it marked invalid. The repair stored a new item metadata stream and new archive metadata, which the index/ fragments do not have. The marker makes the next use rebuild the index from the packs, which have them. """ delete_first_item_chunk(archiver) - real_build = archive_module.build_chunkindex_from_repo - def build_chunkindex_from_repo(repository, **kwargs): - if kwargs.get("write_immediately"): # the rebuild in finish() - raise Error("stopped in the index rebuild") - return real_build(repository, **kwargs) + def write_chunkindex_to_repo(repository, chunks, **kwargs): + raise Error("stopped in the index store") with monkeypatch.context() as m: - m.setattr(archive_module, "build_chunkindex_from_repo", build_chunkindex_from_repo) + m.setattr(archive_module, "write_chunkindex_to_repo", write_chunkindex_to_repo) with open_repository(archiver) as repository: - with pytest.raises(Error, match="stopped in the index rebuild"): + with pytest.raises(Error, match="stopped in the index store"): ArchiveChecker().check(repository, repair=True, sort_by="ts", format="{archive}") with open_repository(archiver) as repository: @@ -1087,7 +1085,7 @@ def test_repair_finish_flushes_pack_writer(archivers, request): checker.key = checker.make_key(repository) checker.repo_objs = RepoObj(checker.key) checker.manifest = Manifest.load(repository, key=checker.key) - # re-adding a chunk makes the chunks index no longer match the packs, so finish() rebuilds it. + checker.chunks = repository.chunks checker.chunks_modified = True # a chunk re-added during repair, buffered in the pack writer: @@ -1099,6 +1097,308 @@ def test_repair_finish_flushes_pack_writer(archivers, request): assert not repository._pack_writer._pieces # finish() stored it +def record_finish_walks(monkeypatch): + """Return a list that collects the id of every pack whose object headers finish() walks. + + ArchiveChecker.verify_written_packs walks a pack with PackReader.iter_headers. + """ + walked = [] + in_finish = False + real_finish = ArchiveChecker.finish + real_iter_headers = PackReader.iter_headers + + def finish(self): + nonlocal in_finish + in_finish = True + try: + return real_finish(self) + finally: + in_finish = False + + def iter_headers(self, *args, **kwargs): + if in_finish: + walked.append(self.pack_id) + return real_iter_headers(self, *args, **kwargs) + + monkeypatch.setattr(ArchiveChecker, "finish", finish) + monkeypatch.setattr(PackReader, "iter_headers", iter_headers) + return walked + + +def list_packs(archiver): + with Repository(archiver.repository_location, exclusive=True) as repository: + return {info.name for info in repository.store_list("packs")} + + +def put_objects_in_one_pack(archiver, contents): + """Store an encrypted repo object per contents entry, all in one new pack no archive references. + + Returns the object ids, in pack order, and the pack id. + """ + with Repository(archiver.repository_location, exclusive=True) as repository: + manifest = Manifest.load(repository) + ids = [] + for data in contents: + chunk_id = manifest.key.id_hash(data) + repository.put(chunk_id, manifest.repo_objs.format(chunk_id, {}, data, ro_type=ROBJ_FILE_STREAM)) + ids.append(chunk_id) + repository.flush() + entries = [repository.chunks[chunk_id] for chunk_id in ids] + assert {entry.pack_id for entry in entries} == {entries[0].pack_id} + assert [entry.obj_offset for entry in entries] == sorted(entry.obj_offset for entry in entries) + return ids, entries[0].pack_id + + +def test_repair_finish_reads_only_the_rewritten_pack(archiver, monkeypatch): + """--verify-data --repair removes a defect chunk; finish() re-reads only the pack delete() wrote.""" + # local-only: this patches in-process archive and repository internals. + monkeypatch.setenv("BORG_PACK_MAX_COUNT", "2") # many packs, so a full walk would be noticed + check_cmd_setup(archiver) + # a defect chunk that no archive references, so the check after the repair finds nothing missing. + # delete() rewrites its pack, keeping the bystander. + (bystander_id, defect_id), pack_id = put_objects_in_one_pack(archiver, [b"bystander", b"defect"]) + with Repository(archiver.repository_location, exclusive=True) as repository: + corrupt_chunk_on_disk(repository, defect_id) + packs_before = list_packs(archiver) + assert len(packs_before) > 10 + + walked = record_finish_walks(monkeypatch) + # the BUFFER_SIZE check_cmd_setup used: rebuild_archives re-chunks the item metadata into the same + # chunks, so it stores nothing and the rewritten pack is the only pack the repair writes. + with patch.object(ChunkBuffer, "BUFFER_SIZE", 10): + output = cmd(archiver, "check", "--repair", "--verify-data", exit_code=0) + assert f"{bin_to_hex(defect_id)}, integrity error" in output + + new_packs = list_packs(archiver) - packs_before + assert packs_before - list_packs(archiver) == {bin_to_hex(pack_id)} + assert len(new_packs) == 1 + assert [bin_to_hex(pack_id) for pack_id in walked] == list(new_packs) + with Repository(archiver.repository_location, exclusive=True) as repository: + assert defect_id not in repository.chunks + assert bin_to_hex(repository.chunks[bystander_id].pack_id) in new_packs + cmd(archiver, "check", exit_code=0) + + +def test_repair_finish_reads_no_pack_after_deleting_a_whole_pack(archiver, monkeypatch): + """--verify-data --repair removes a defect chunk that is alone in its pack; finish() re-reads no pack. + + delete() drops the whole pack and writes no new one, so the repair wrote no pack. + """ + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + (defect_id,), pack_id = put_objects_in_one_pack(archiver, [b"defect"]) + with Repository(archiver.repository_location, exclusive=True) as repository: + corrupt_chunk_on_disk(repository, defect_id) + packs_before = list_packs(archiver) + + walked = record_finish_walks(monkeypatch) + findings = record_verify_findings(monkeypatch) + with patch.object(ChunkBuffer, "BUFFER_SIZE", 10): # see test_repair_finish_reads_only_the_rewritten_pack + output = cmd(archiver, "check", "--repair", "--verify-data", "--info", exit_code=0) + assert f"{bin_to_hex(defect_id)}, integrity error" in output + assert findings == [False] + assert "Re-reading the packs written by the repair" not in output + assert walked == [] + assert list_packs(archiver) == packs_before - {bin_to_hex(pack_id)} + with Repository(archiver.repository_location, exclusive=True) as repository: + assert defect_id not in repository.chunks + cmd(archiver, "check", exit_code=0) + + +def test_repair_finish_reads_only_the_packs_put_wrote(archiver, monkeypatch): + """--repair re-stores a missing item metadata chunk; finish() re-reads only the packs put() wrote.""" + # local-only: this patches in-process archive and repository internals. + # a pack per object, so every put() after the first returns the pack the background store-thread + # stored before it. + monkeypatch.setenv("BORG_PACK_MAX_COUNT", "1") + check_cmd_setup(archiver) + archive, repository = open_archive(archiver.repository_path, "archive1") + with repository: + repository.delete(archive.item_ids[0], validate=None) + packs_before = list_packs(archiver) + + walked = record_finish_walks(monkeypatch) + findings = record_verify_findings(monkeypatch) + cmd(archiver, "check", "--repair", exit_code=0) + assert findings == [False] + + new_packs = list_packs(archiver) - packs_before + assert new_packs + assert sorted(bin_to_hex(pack_id) for pack_id in walked) == sorted(new_packs) # each one once + cmd(archiver, "check", exit_code=0) + + +def test_repair_finish_reads_the_pack_its_flush_stores(archiver, monkeypatch): + """finish() re-reads the pack its own flush stores, e.g. for chunks buffered when a Ctrl-C stopped the repair.""" + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + walked = record_finish_walks(monkeypatch) + with Repository(archiver.repository_location, exclusive=True) as repository: + checker = ArchiveChecker() + checker.repair = True + checker.repository = repository + checker.key = checker.make_key(repository) + checker.repo_objs = RepoObj(checker.key) + checker.manifest = Manifest.load(repository, key=checker.key) + checker.chunks = repository.chunks + checker.chunks_modified = True + data = b"repaired" + chunk_id = checker.key.id_hash(data) + assert repository.put(chunk_id, checker.repo_objs.format(chunk_id, {}, data, ro_type=ROBJ_FILE_STREAM)) is None + checker.finish() + assert not checker.error_found + with Repository(archiver.repository_location, exclusive=True) as repository: + assert walked == [repository.chunks[chunk_id].pack_id] + + +def test_repair_finish_reads_a_rewritten_pack_no_index_entry_names(archiver, monkeypatch): + """finish() re-reads a pack delete() wrote, also when no index entry names that pack. + + The pack holds an object with a corrupt header, which the rebuild in check() drops, and a defect + chunk, which --verify-data --repair deletes. compact_pack copies the dropped object's bytes (no + index entry covers them) into the new pack, so the new pack exists, but no index entry points at it. + """ + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + (dropped_id, defect_id), pack_id = put_objects_in_one_pack(archiver, [b"dropped", b"defect"]) + with Repository(archiver.repository_location, exclusive=True) as repository: + corrupt_chunk_on_disk(repository, defect_id) # the payload: the header still validates + key = "packs/" + bin_to_hex(pack_id) + dropped = repository.chunks[dropped_id] + repository.store_store(key, corrupt(repository.store_load(key), dropped.obj_offset)) # the magic + packs_before = list_packs(archiver) + + walked = record_finish_walks(monkeypatch) + with patch.object(ChunkBuffer, "BUFFER_SIZE", 10): # see test_repair_finish_reads_only_the_rewritten_pack + output = cmd(archiver, "check", "--archives-only", "--repair", "--verify-data", "--debug", exit_code=0) + assert "no object header at offset 0" in output + assert f"{bin_to_hex(defect_id)}, integrity error" in output + + assert packs_before - list_packs(archiver) == {bin_to_hex(pack_id)} + new_packs = list_packs(archiver) - packs_before + assert len(new_packs) == 1 + assert [bin_to_hex(pack_id) for pack_id in walked] == list(new_packs) + with Repository(archiver.repository_location, exclusive=True) as repository: + assert not any(bin_to_hex(entry.pack_id) in new_packs for _, entry in repository.chunks.iteritems()) + + +def test_repair_finish_accepts_a_superseded_duplicate_in_a_rewritten_pack(archiver, monkeypatch): + """A superseded duplicate that delete() copies into the new pack is not a finding of finish(). + + The pack holds an object with a corrupt header, two copies of one chunk and a defect chunk. The + rebuild in check() drops the first object and indexes the second copy. compact_pack copies the bytes + before the second copy, which no index entry covers, into the new pack: its search for superseded + duplicates there stops at the corrupt header. So the new pack holds both copies, the index names + only the second. + """ + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + monkeypatch.setenv("BORG_PACK_MAX_COUNT", "4") # the four objects below go into one pack + (dropped_id, dup_id, _, defect_id), pack_id = put_objects_in_one_pack( + archiver, [b"dropped", b"duplicate", b"duplicate", b"defect"] + ) + with Repository(archiver.repository_location, exclusive=True) as repository: + corrupt_chunk_on_disk(repository, defect_id) # the payload: the header still validates + key = "packs/" + bin_to_hex(pack_id) + dropped = repository.chunks[dropped_id] + repository.store_store(key, corrupt(repository.store_load(key), dropped.obj_offset)) # the magic + + walked = record_finish_walks(monkeypatch) + with patch.object(ChunkBuffer, "BUFFER_SIZE", 10): # see test_repair_finish_reads_only_the_rewritten_pack + output = cmd(archiver, "check", "--archives-only", "--repair", "--verify-data", "--debug", exit_code=0) + assert f"{bin_to_hex(defect_id)}, integrity error" in output + assert "in a gap, keeping the remaining" in output + assert len(walked) == 1 + assert "the chunks index does not match the pack" not in output + with Repository(archiver.repository_location, exclusive=True) as repository: + entry = repository.chunks[dup_id] + assert entry.pack_id == walked[0] + # the second copy: the first one starts where the dropped object ends. + assert entry.obj_offset > dropped.obj_size + cmd(archiver, "check", exit_code=0) + + +def record_verify_findings(monkeypatch, tamper=None): + """Return a list that collects, per verify_written_packs call, whether that call found a problem. + + tamper(checker) runs right before the call. The problems found before it (e.g. the damage the + repair fixed) stay recorded in checker.error_found, but do not count for the call. + """ + findings = [] + real_verify = ArchiveChecker.verify_written_packs + + def verify_written_packs(self): + if tamper is not None: + tamper(self) + error_found, self.error_found = self.error_found, False + try: + return real_verify(self) + finally: + findings.append(self.error_found) + self.error_found = self.error_found or error_found + + monkeypatch.setattr(ArchiveChecker, "verify_written_packs", verify_written_packs) + return findings + + +def test_repair_finish_fixes_a_wrong_index_entry_for_a_written_pack(archiver, monkeypatch): + """finish() compares the written packs with their index entries, reports a difference and fixes it.""" + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + archive, repository = open_archive(archiver.repository_path, "archive1") + with repository: + repository.delete(archive.item_ids[0], validate=None) + + tampered = {} + + def tamper(checker): + # an index entry with a wrong offset, as a bug in the offset arithmetic would make one. + pack_id = min(checker.written_packs) + chunk_id, entry = next((cid, e) for cid, e in checker.chunks.iteritems() if e.pack_id == pack_id) + checker.chunks[chunk_id] = entry._replace(obj_offset=entry.obj_offset + 1) + tampered[chunk_id] = entry + + findings = record_verify_findings(monkeypatch, tamper) + output = cmd(archiver, "check", "--repair", exit_code=0) + assert findings == [True] + ((chunk_id, entry),) = tampered.items() + assert f"pack {bin_to_hex(entry.pack_id)}: the chunks index does not match the pack" in output + assert "Indexed objects not in the pack: 1, objects in the pack with an unindexed chunk id: 1." in output + assert "Archive consistency check complete, problems found." in output + + with Repository(archiver.repository_location, exclusive=True) as repository: + stored = repository.chunks[chunk_id] + assert (stored.pack_id, stored.obj_offset, stored.obj_size) == (entry.pack_id, entry.obj_offset, entry.obj_size) + cmd(archiver, "check", exit_code=0) + + +def test_repair_finish_reports_a_missing_written_pack(archiver, monkeypatch): + """finish() reports a written pack that is gone and removes the index entries that name it.""" + # local-only: this patches in-process archive and repository internals. + check_cmd_setup(archiver) + archive, repository = open_archive(archiver.repository_path, "archive1") + with repository: + repository.delete(archive.item_ids[0], validate=None) + + removed = [] + + def tamper(checker): + # a written pack that vanished, as a store losing it would make it. + pack_id = min(checker.written_packs) + checker.repository.store_delete("packs/" + bin_to_hex(pack_id)) + removed.append(pack_id) + + findings = record_verify_findings(monkeypatch, tamper) + output = cmd(archiver, "check", "--repair", exit_code=0) + assert findings == [True] + (pack_id,) = removed + assert f"pack {bin_to_hex(pack_id)}: written by the repair, but it is missing." in output + assert "the chunks index does not match the pack" not in output + assert "Archive consistency check complete, problems found." in output + with Repository(archiver.repository_location, exclusive=True) as repository: + assert not any(entry.pack_id == pack_id for _, entry in repository.chunks.iteritems()) + + @pytest.mark.parametrize("init_args", [["--encryption=aes256-ocb"], ["--encryption", "none-sha256"]]) def test_verify_data(archivers, request, init_args): archiver = request.getfixturevalue(archivers) diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index b56b957455..95eeb1f9de 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -410,6 +410,18 @@ def test_read_data(repo_fixtures, request): assert repository.get(H(0), read_data=False) == chunk_short +def test_flush_returns_the_stored_objects(repository): + assert repository.flush() is None # not opened, no pack writer + with repository: + assert repository.flush() is None # nothing buffered + repository.put(H(0), fchunk(b"foo")) + ((chunk_id, pack_id, obj_offset, obj_size),) = repository.flush() + entry = repository.chunks[H(0)] + assert (chunk_id, pack_id, obj_offset, obj_size) == (H(0), entry.pack_id, entry.obj_offset, entry.obj_size) + assert repository.flush() is None + assert repository.flush() is None # closed + + def test_consistency(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: repository.put(H(0), fchunk(b"foo"))