From 5e4f8f8749067c4f430463039cdf8b122c49104d Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Thu, 1 Oct 2026 09:44:09 +0530 Subject: [PATCH 1/3] compact, repo-compress: keep a superseded duplicate indexed in a missing or corrupt pack, fixes #10474 --- docs/internals/packs.rst | 4 +- src/borg/archiver/compact_cmd.py | 10 +++- src/borg/archiver/repo_compress_cmd.py | 2 + src/borg/repository.py | 47 ++++++++++++++----- .../testsuite/archiver/compact_cmd_test.py | 44 ++++++++++++++++- .../archiver/repo_compress_cmd_test.py | 28 +++++++++++ src/borg/testsuite/repository_test.py | 12 +++++ 7 files changed, 132 insertions(+), 15 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 381cf2eab8..906e82876e 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -329,7 +329,9 @@ Rewriting a pack (``compact_pack``, ``transform_pack``) drops the superseded gap header and metadata slot validate, checked as in the repair walk above (``repoobj.object_validator``), and copies all other gap bytes into the new pack. Validation covers ``meta_size`` and ``data_size``, so a dropped range is exactly one blob. -Without a validator (``validate=None``), no gap bytes are dropped. Merging packs +Without a validator (``validate=None``), no gap bytes are dropped. ``borg compact`` and +``borg repo-compress`` also keep a superseded blob whose indexed copy is in a pack missing +from the store or recorded corrupt, as that copy may be unreadable. Merging packs (``merge_packs``) copies whole pack files, so it keeps all gap bytes. The walk over a gap steps from header to header by the blob size each header states. It ends diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 9c69db8955..ac7ab0d541 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -299,11 +299,13 @@ def compact_packs(self): pack_used = defaultdict(int) pack_indexed = defaultdict(int) # pack_id -> bytes of all its index entries, used or not stale_ids = [] # index entries referencing a pack file that is not in the store + stale_packs = set() # ids of those missing pack files stale_used = 0 # how many of those were still flagged used (lost data) for id, entry in self.chunks.iteritems(): pid = entry.pack_id if pid not in pack_total: stale_ids.append(id) + stale_packs.add(pid) if entry.flags & ChunkIndex.F_USED: stale_used += 1 else: @@ -455,13 +457,19 @@ def compact_packs(self): progress += 1 pi.show(progress) # report after the work, so the final pack lands on 100% validate = object_validator(self.manifest.repo_objs) + untrusted_pack_ids = stale_packs | corrupt_packs for pid in rewrite_packs: if sig_int: break # chunks=self.chunks: the index updates (repoint kept objects, remove dropped ones) # must land in the index that save_chunk_index() persists (#9850). _, dropped = self.repository.compact_pack( - pid, keep_ids=keep[pid], drop_ids=drop[pid], chunks=self.chunks, validate=validate + pid, + keep_ids=keep[pid], + drop_ids=drop[pid], + chunks=self.chunks, + validate=validate, + untrusted_pack_ids=untrusted_pack_ids, ) freed += dropped # unused indexed objects plus superseded duplicates progress += 1 diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py index 91b1d9fc6c..294fc51776 100644 --- a/src/borg/archiver/repo_compress_cmd.py +++ b/src/borg/archiver/repo_compress_cmd.py @@ -120,6 +120,7 @@ def recompress(self): total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress" ) validate = object_validator(self.repo_objs) + untrusted_pack_ids = stale_packs | corrupt_packs for i, (pack_id, pack_size) in enumerate(packs): if sig_int: break # stop cleanly at a pack boundary: save the index below, then raise @@ -133,6 +134,7 @@ def recompress(self): chunks=self.chunks, before_change=self.invalidate_stored_index, validate=validate, + untrusted_pack_ids=untrusted_pack_ids, ) if new_pack_id != pack_id: self.packs_rewritten += 1 diff --git a/src/borg/repository.py b/src/borg/repository.py index 6ce95378f0..781d28d598 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -595,18 +595,19 @@ def check_pack_objects(pack_hex, obj_ranges, pack_size): ) -def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, validate): +def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, validate, untrusted_pack_ids=frozenset()): """Return the offset-ordered (offset, size) ranges of the superseded duplicates in a pack's gaps. A gap is a byte range of the pack that no chunks index entry covers. A superseded duplicate is - an object in a gap whose chunk id the index maps to another location. Equal chunk ids mean - equal plaintext, so its bytes are redundant, whatever the stored size of the indexed copy. + an object in a gap whose chunk id the index maps to another location, in a pack not in + untrusted_pack_ids. Equal chunk ids mean equal plaintext, so its bytes are redundant, whatever + the stored size of the indexed copy. Each gap is walked from object header to object header, stepping by the object size the header - states. An object is reported when its chunk id is indexed at another location and validate - accepts it. The walk over a gap ends at a header that does not parse or that reaches past the - gap. Objects validate rejects are kept, and so is the rest of a gap where the walk ends early; - both are logged as a warning with the pack id and the offset. + states. An object is reported when it is a superseded duplicate and validate accepts it. The + walk over a gap ends at a header that does not parse or that reaches past the gap. Objects + validate rejects are kept, and so is the rest of a gap where the walk ends early; both are + logged as a warning with the pack id and the offset. reader: PackReader of the pack. chunks: the chunks index (chunk id -> ChunkIndexEntry). @@ -618,6 +619,8 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val header and metadata slot (the meta_size metadata bytes after the header). True means chunk id, meta_size and data_size are verified, so the reported range is exactly the object. With None, nothing is reported. + untrusted_pack_ids: ids of packs whose objects may be unreadable, e.g. packs missing from the + store or recorded corrupt. A gap object whose chunk id the index maps into one of them is kept. """ if validate is None: return [] @@ -651,7 +654,11 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val break obj_size = hdr_size + hdr.meta_size + hdr.data_size entry = chunks.get(hdr.chunk_id) - if entry is not None and (entry.pack_id != pack_id or entry.obj_offset != offset): + if ( + entry is not None + and entry.pack_id not in untrusted_pack_ids + and (entry.pack_id != pack_id or entry.obj_offset != offset) + ): problem = reader._validation_problem(hdr, offset, buf, offset, validate) if problem is None: drop_ranges.append((offset, obj_size)) @@ -2141,7 +2148,15 @@ def delete(self, id, *, validate, update_index=True): return result def compact_pack( - self, pack_id, *, keep_ids: set, drop_ids: set, validate, chunks=None, before_old_pack_delete=None + self, + pack_id, + *, + keep_ids: set, + drop_ids: set, + validate, + chunks=None, + before_old_pack_delete=None, + untrusted_pack_ids=frozenset(), ): """Rewrite pack , keeping and dropping , then delete the old pack. @@ -2152,6 +2167,7 @@ def compact_pack( updates to. Must be the index keep_ids and drop_ids were derived from. Default: self.chunks. before_old_pack_delete: callable without arguments, called once just before the old pack is deleted. Not called when no bytes are dropped, since the old pack then stays. + untrusted_pack_ids: passed to superseded_gap_ranges. Together, keep_ids and drop_ids must cover every object the chunk index lists for this pack; an unlisted indexed object would keep its bytes in the new pack but its index entry would go @@ -2199,7 +2215,9 @@ def compact_pack( # toward the rewrite threshold and a wholly superseded orphan pack can be dropped outright. drop_ranges = [(offset, size) for offset, _, size, keep in located if not keep] reader = PackReader(store=self.store, pack_id=pack_id) - drop_ranges += superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, validate=validate) + drop_ranges += superseded_gap_ranges( + reader, chunks, pack_id, obj_ranges, pack_size, validate=validate, untrusted_pack_ids=untrusted_pack_ids + ) drop_ranges.sort() dropped_bytes = sum(size for _, size in drop_ranges) # on-disk bytes this rewrite frees, for --stats @@ -2358,7 +2376,9 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): pi.show(increase=1) pi.finish() - def transform_pack(self, pack_id, ids, transform, *, validate, chunks=None, before_change=None): + def transform_pack( + self, pack_id, ids, transform, *, validate, chunks=None, before_change=None, untrusted_pack_ids=frozenset() + ): """Rewrite pack , passing each indexed object's bytes through . ids: the chunk ids of this pack's objects. Must cover every object the chunk index lists @@ -2373,6 +2393,7 @@ def transform_pack(self, pack_id, ids, transform, *, validate, chunks=None, befo updates to. Must be the index was derived from. Default: self.chunks. before_change: called once, just before the first store modification; use it to invalidate stored chunk indexes for crash safety (see #9748). Not called when the pack is kept. + untrusted_pack_ids: passed to superseded_gap_ranges. The whole pack file is loaded into memory (bounded by the pack size limit). Gaps (byte ranges no index entry covers) are copied into the new pack, except the superseded duplicates @@ -2410,7 +2431,9 @@ def transform_pack(self, pack_id, ids, transform, *, validate, chunks=None, befo located.sort() obj_ranges = [(offset, size) for offset, _, size in located] check_pack_objects(pack_hex, obj_ranges, pack_size) - drop_ranges = superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, validate=validate) + drop_ranges = superseded_gap_ranges( + reader, chunks, pack_id, obj_ranges, pack_size, validate=validate, untrusted_pack_ids=untrusted_pack_ids + ) # assemble the new pack in offset order: transformed objects, dropped ranges skipped, all # other bytes copied verbatim. the two range lists never overlap (drops lie in gaps), so a diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index d94098d01c..ac8a6e2991 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -9,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, PackTracker +from ...repository import Repository, PackReader, 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 @@ -314,6 +314,48 @@ def test_compact_superseded_duplicate(tmp_path): assert new_size == pack_a_size - y_size - x_size +@pytest.mark.parametrize("untrusted", ("missing", "corrupt")) +def test_compact_keeps_duplicate_indexed_in_untrusted_pack(tmp_path, untrusted): + # X is indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from the + # store or recorded corrupt, the copy in pack A may be the only readable one: rewriting pack A + # (for its unused Y) keeps it (#10474). + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + w, x, y = b"WWWW", b"XXXX", os.urandom(1000) + w_id, x_id, y_id = (repo_objs.id_hash(data) for data in (w, x, y)) + repository._pack_writer.max_count = 4 # one flush() -> one pack + for cid, data in [(w_id, w), (x_id, x), (y_id, y)]: + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_a = repository.chunks[w_id].pack_id + pack_a_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(pack_a)) + y_size = repository.chunks[y_id].obj_size + repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_b = repository.chunks[x_id].pack_id + assert pack_b != pack_a + if untrusted == "missing": + repository.store_delete("packs/" + bin_to_hex(pack_b)) + else: + record_corrupt(repository, pack_b) + for cid in (w_id, x_id, y_id): + flags = ChunkIndex.F_NONE if cid == y_id else ChunkIndex.F_USED + repository.chunks[cid] = repository.chunks[cid]._replace(flags=flags) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + new_pack = repository.chunks[w_id].pack_id + assert new_pack != pack_a # pack A was rewritten + new_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(new_pack)) + assert new_size == pack_a_size - y_size # only the unused Y was dropped + reader = PackReader(store=repository.store, pack_id=new_pack) + assert x_id in [chunk_id for chunk_id, _, _ in reader.iter_headers()] # the gap copy of X is kept + + def test_compact_keeps_orphan_pack(tmp_path): # A pack with no index entries at all (a backup that crashed after writing the pack but before # writing its index) is left alone: its objects hold no reclaimable indexed waste and may be live diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 8bb2e3f3b4..9b3eaed8ce 100644 --- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py +++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py @@ -349,6 +349,34 @@ def test_transform_pack_drops_superseded_gap(tmp_path): assert pdchunk(repository.get(H(1))) == b"XXXX" # the authoritative copy in pack B +def test_repo_compress_keeps_duplicate_indexed_in_missing_pack(archiver): + # X is indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from the + # store, the copy in pack A is the only one left: repo-compress keeps it (#10474). + cmd(archiver, "repo-create", RK_ENCRYPTION) + with open_repository(archiver) as repository: + repo_objs = Manifest.load(repository).repo_objs + w, x = b"W" * 1000, b"X" * 1000 + w_id, x_id = repo_objs.id_hash(w), repo_objs.id_hash(x) + repository._pack_writer.max_count = 2 # one flush() -> one pack + for cid, data in [(w_id, w), (x_id, x)]: + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_a = repository.chunks[w_id].pack_id + repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_b = repository.chunks[x_id].pack_id + assert pack_b != pack_a + repository.store_delete("packs/" + bin_to_hex(pack_b)) + + cmd(archiver, "repo-compress", "-C", "none") + + with open_repository(archiver) as repository: + new_pack = repository.chunks[w_id].pack_id + assert new_pack != pack_a # W was recompressed, pack A was rewritten + reader = PackReader(store=repository.store, pack_id=new_pack) + assert x_id in [chunk_id for chunk_id, _, _ in reader.iter_headers()] # the gap copy of X is kept + + def test_transform_pack_unchanged_pack_untouched(tmp_path): # if every transform keeps its object, the store must not be touched at all: # no pack write, no pack delete, no before_change call. diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 5a93e94fde..0c7edfd062 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -3282,6 +3282,18 @@ def test_superseded_gap_ranges_reports_an_authenticated_duplicate(tmp_path): assert gap_ranges(obj, chunks, None) == [] # no validator, nothing to report +def test_superseded_gap_ranges_keeps_a_duplicate_indexed_in_an_untrusted_pack(tmp_path): + repo_objs = aead_repo_objs(tmp_path) + (obj,), chunks = gap_pack(repo_objs, [b"superseded"]) + reader = PackReader(pack_contents=obj) + validate = object_validator(repo_objs) + + ranges = superseded_gap_ranges( + reader, chunks, THIS_PACK, [], len(obj), validate=validate, untrusted_pack_ids={OTHER_PACK} + ) + assert ranges == [] + + def test_superseded_gap_ranges_rejects_a_forged_chunk_id(tmp_path): # the header's chunk id is replaced by the id of another indexed chunk, which the metadata # slot's tag does not authenticate. From e84e8f9095708ebc647d3f62074b056b412389cf Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sun, 4 Oct 2026 22:43:31 +0530 Subject: [PATCH 2/3] check, compact, repo-compress: keep a superseded duplicate indexed in a pack with a defect chunk or in a truncated pack --- docs/internals/packs.rst | 10 ++-- src/borg/archive.py | 10 +++- src/borg/archiver/compact_cmd.py | 5 +- src/borg/archiver/repo_compress_cmd.py | 16 ++++-- src/borg/repository.py | 7 ++- src/borg/testsuite/archiver/check_cmd_test.py | 43 +++++++++++++++ .../testsuite/archiver/compact_cmd_test.py | 24 +++++---- .../archiver/repo_compress_cmd_test.py | 52 +++++++++++++------ 8 files changed, 129 insertions(+), 38 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 906e82876e..d0581f9e50 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -329,10 +329,12 @@ Rewriting a pack (``compact_pack``, ``transform_pack``) drops the superseded gap header and metadata slot validate, checked as in the repair walk above (``repoobj.object_validator``), and copies all other gap bytes into the new pack. Validation covers ``meta_size`` and ``data_size``, so a dropped range is exactly one blob. -Without a validator (``validate=None``), no gap bytes are dropped. ``borg compact`` and -``borg repo-compress`` also keep a superseded blob whose indexed copy is in a pack missing -from the store or recorded corrupt, as that copy may be unreadable. Merging packs -(``merge_packs``) copies whole pack files, so it keeps all gap bytes. +Without a validator (``validate=None``), no gap bytes are dropped. A superseded blob is +also kept when its indexed copy may be unreadable: ``borg compact`` and ``borg repo-compress`` +keep it when that copy is in a pack that is missing from the store, recorded corrupt, or +shorter than its index entries state, ``borg check --repair --verify-data`` when that copy is +in a missing pack or in a pack holding a defect chunk. Merging packs (``merge_packs``) copies +whole pack files, so it keeps all gap bytes. The walk over a gap steps from header to header by the blob size each header states. It ends at a header that does not parse or that reaches past the gap. The rest of that gap is kept, diff --git a/src/borg/archive.py b/src/borg/archive.py index 0a0c548102..51d2080f90 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -2503,6 +2503,7 @@ def verify_data(self): verified = 0 # chunks actually verified defect_chunks = [] archive_meta_ids = set() + untrusted_pack_ids = set() # ids of the packs that are missing or hold a defect chunk pi = ProgressIndicatorPercent( total=chunks_count, msg="Verifying data %6.2f%%", step=0.01, msgid="check.verify_data" ) @@ -2533,11 +2534,13 @@ def verify_data(self): errors += 1 logger.error("chunk %s, integrity error: %s", bin_to_hex(chunk_id), integrity_error) defect_chunks.append(chunk_id) + untrusted_pack_ids.add(pack_id) except Repository.PackNotFound: # the pack is gone, thus every chunk the index places in it is lost. get_many() loads the # whole pack for the first chunk, so it raises before any chunk of this pack was read. # one error line for the pack, the chunk ids at debug level: a pack holds thousands of them. self.error_found = True + untrusted_pack_ids.add(pack_id) lost = len(chunk_ids) errors += lost verified += lost # they are not read, but they are accounted for, like a failed read @@ -2580,13 +2583,18 @@ def verify_data(self): # 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 # and clears the invalid marker delete() writes. + # untrusted_pack_ids: the rewrite keeps an object of the pack that no index entry + # covers, if the index maps its chunk id into one of these packs. old_pack_id = self.chunks[defect_chunk].pack_id # new_pack_id: the pack holding the other objects of the old pack, None if there were none. - new_pack_id, _ = self.repository.delete(defect_chunk, update_index=False, validate=validate) + new_pack_id, _ = self.repository.delete( + defect_chunk, update_index=False, validate=validate, untrusted_pack_ids=untrusted_pack_ids + ) self.chunks_modified = True self.written_packs.discard(old_pack_id) # delete() removed the old pack if new_pack_id is not None: self.written_packs.add(new_pack_id) + untrusted_pack_ids.add(new_pack_id) # it may hold other defect chunks of the old pack else: logger.warning("chunk %s not deleted, did not consistently fail.", bin_to_hex(defect_chunk)) if meta["type"] == ROBJ_ARCHIVE_META: diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index ac7ab0d541..ce44c19493 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -300,6 +300,7 @@ def compact_packs(self): pack_indexed = defaultdict(int) # pack_id -> bytes of all its index entries, used or not stale_ids = [] # index entries referencing a pack file that is not in the store stale_packs = set() # ids of those missing pack files + short_packs = set() # ids of the pack files that end before one of their index entries does stale_used = 0 # how many of those were still flagged used (lost data) for id, entry in self.chunks.iteritems(): pid = entry.pack_id @@ -310,6 +311,8 @@ def compact_packs(self): stale_used += 1 else: pack_indexed[pid] += entry.obj_size + if entry.obj_offset + entry.obj_size > pack_total[pid]: + short_packs.add(pid) if entry.flags & ChunkIndex.F_USED: pack_used[pid] += entry.obj_size @@ -457,7 +460,7 @@ def compact_packs(self): progress += 1 pi.show(progress) # report after the work, so the final pack lands on 100% validate = object_validator(self.manifest.repo_objs) - untrusted_pack_ids = stale_packs | corrupt_packs + untrusted_pack_ids = stale_packs | corrupt_packs | short_packs for pid in rewrite_packs: if sig_int: break diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py index 294fc51776..449a1c2542 100644 --- a/src/borg/archiver/repo_compress_cmd.py +++ b/src/borg/archiver/repo_compress_cmd.py @@ -82,15 +82,21 @@ def recompress(self): logger.info(f"Recompressing repository to {format_compression_spec(*self.wanted)}...") self.chunks = build_chunkindex_from_repo(self.repository, write_immediately=True) + # the pack files actually in the store; sorted, so the processing order is reproducible. + packs = sorted( + (hex_to_bin(info.name), info.size) for info in map(ItemInfo._make, self.repository.store_list("packs")) + ) + pack_sizes = dict(packs) + # group the indexed objects per pack; transform_pack requires each pack's complete id list. per_pack = defaultdict(list) # pack_id -> [chunk_id, ...] + short_packs = set() # ids of the pack files that end before one of their index entries does for id, entry in self.chunks.iteritems(): per_pack[entry.pack_id].append(id) + pack_size = pack_sizes.get(entry.pack_id) + if pack_size is not None and entry.obj_offset + entry.obj_size > pack_size: + short_packs.add(entry.pack_id) - # the pack files actually in the store; sorted, so the processing order is reproducible. - packs = sorted( - (hex_to_bin(info.name), info.size) for info in map(ItemInfo._make, self.repository.store_list("packs")) - ) self.packs_count = len(packs) size_before = sum(size for _, size in packs) size_after = size_before @@ -120,7 +126,7 @@ def recompress(self): total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress" ) validate = object_validator(self.repo_objs) - untrusted_pack_ids = stale_packs | corrupt_packs + untrusted_pack_ids = stale_packs | corrupt_packs | short_packs for i, (pack_id, pack_size) in enumerate(packs): if sig_int: break # stop cleanly at a pack boundary: save the index below, then raise diff --git a/src/borg/repository.py b/src/borg/repository.py index 781d28d598..730fd25a2d 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -620,7 +620,8 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val id, meta_size and data_size are verified, so the reported range is exactly the object. With None, nothing is reported. untrusted_pack_ids: ids of packs whose objects may be unreadable, e.g. packs missing from the - store or recorded corrupt. A gap object whose chunk id the index maps into one of them is kept. + store, recorded corrupt, or shorter than their index entries state. A gap object whose chunk + id the index maps into one of them is kept. """ if validate is None: return [] @@ -2105,7 +2106,7 @@ def put(self, id, data): # PackWriter shares this repository's index, so add() triggers the lazy build itself. return self._pack_writer.add(id, data) - def delete(self, id, *, validate, update_index=True): + def delete(self, id, *, validate, update_index=True, untrusted_pack_ids=frozenset()): """Delete a single repo object by rewriting its pack without it (via compact_pack). The rewrite deletes the old pack, so the index/ fragments point the pack's other objects at a @@ -2118,6 +2119,7 @@ def delete(self, id, *, validate, update_index=True): 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. + untrusted_pack_ids: 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. @@ -2139,6 +2141,7 @@ def delete(self, id, *, validate, update_index=True): drop_ids={id}, validate=validate, before_old_pack_delete=lambda: write_chunkindex_invalid(self), + untrusted_pack_ids=untrusted_pack_ids, ) if update_index: # close() only persists new entries incrementally, so write the full index here to record diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 6ebac7ea96..e6726a7404 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -20,6 +20,7 @@ list_chunkindex_hashes, read_chunkindex_from_repo, write_chunkindex_invalid, + write_chunkindex_to_repo, ) from ...crypto.key import RepositoryKeyInfoMissing from ...constants import * # NOQA @@ -1674,6 +1675,48 @@ def test_repair_finish_accepts_a_superseded_duplicate_in_a_rewritten_pack(archiv cmd(archiver, "check", exit_code=0) +def test_verify_data_repair_keeps_gap_copy_of_a_defect_chunk(archiver): + """--verify-data --repair keeps a gap object whose chunk id the index maps to a defect chunk. + + Pack A holds a defect chunk D and an intact copy of X that no index entry covers. The index maps X + to pack B, where X is defect. Removing D rewrites pack A and keeps its copy of X, which finish() + indexes after the defect X was removed. + """ + cmd(archiver, "repo-create", RK_ENCRYPTION) + with KeyedRepository(archiver.repository_location, exclusive=True) as repository: + repo_objs = Manifest.load(repository).repo_objs + d, w, x = b"defect", b"other", b"duplicate" + d_id, w_id, x_id = (repo_objs.id_hash(data) for data in (d, w, x)) + for cid, data in [(d_id, d), (w_id, w), (x_id, x)]: + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_a = repository.chunks[d_id].pack_id + assert repository.chunks[x_id].pack_id == pack_a + # verify_data removes the defect chunks in pack id order: D must be removed while the index maps X + # to pack B, so pack B needs the higher pack id. The pack id depends on the random nonce of X. + while True: + repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_b = repository.chunks[x_id].pack_id + if pack_b > pack_a: + break + repository.store_delete("packs/" + bin_to_hex(pack_b)) + # a full index, so that the check loads X as indexed in pack B. + write_chunkindex_to_repo(repository, repository.chunks, incremental=False, force_write=True, delete_other=True) + corrupt_chunk_on_disk(repository, d_id) + corrupt_chunk_on_disk(repository, x_id) # the indexed copy, in pack B + + output = cmd(archiver, "check", "--archives-only", "--repair", "--verify-data", exit_code=0) + assert f"{bin_to_hex(d_id)}, integrity error" in output + assert f"{bin_to_hex(x_id)}, integrity error" in output + + with KeyedRepository(archiver.repository_location, exclusive=True) as repository: + repo_objs = Manifest.load(repository).repo_objs + assert d_id not in repository.chunks + assert repository.chunks[x_id].pack_id not in (pack_a, pack_b) + assert repo_objs.parse(x_id, repository.get(x_id), ro_type=ROBJ_FILE_STREAM)[1] == x + + def record_verify_findings(monkeypatch, tamper=None): """Return a list that collects, per verify_written_packs call, whether that call found a problem. diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index ac8a6e2991..e1f7f26ccf 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -6,7 +6,7 @@ import pytest from ...constants import * # NOQA -from ...helpers import get_cache_dir, bin_to_hex, sig_int, Error +from ...helpers import get_cache_dir, bin_to_hex, hex_to_bin, sig_int, Error from ...hashindex import ChunkIndex from ...repoobj import RepoObj from ...repository import Repository, PackReader, PackTracker @@ -314,11 +314,11 @@ def test_compact_superseded_duplicate(tmp_path): assert new_size == pack_a_size - y_size - x_size -@pytest.mark.parametrize("untrusted", ("missing", "corrupt")) +@pytest.mark.parametrize("untrusted", ("missing", "corrupt", "truncated")) def test_compact_keeps_duplicate_indexed_in_untrusted_pack(tmp_path, untrusted): # X is indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from the - # store or recorded corrupt, the copy in pack A may be the only readable one: rewriting pack A - # (for its unused Y) keeps it (#10474). + # store, recorded corrupt or truncated before the end of X, the copy in pack A may be the only + # readable one: rewriting pack A (for its unused Y) keeps it (#10474). location = os.fspath(tmp_path / "repo") with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: manifest = gc_manifest(repository) @@ -336,10 +336,13 @@ def test_compact_keeps_duplicate_indexed_in_untrusted_pack(tmp_path, untrusted): repository.flush() pack_b = repository.chunks[x_id].pack_id assert pack_b != pack_a + key_b = "packs/" + bin_to_hex(pack_b) if untrusted == "missing": - repository.store_delete("packs/" + bin_to_hex(pack_b)) - else: + repository.store_delete(key_b) + elif untrusted == "corrupt": record_corrupt(repository, pack_b) + else: + repository.store_store(key_b, repository.store_load(key_b)[:-1]) for cid in (w_id, x_id, y_id): flags = ChunkIndex.F_NONE if cid == y_id else ChunkIndex.F_USED repository.chunks[cid] = repository.chunks[cid]._replace(flags=flags) @@ -348,11 +351,12 @@ def test_compact_keeps_duplicate_indexed_in_untrusted_pack(tmp_path, untrusted): gc.chunks = repository.chunks gc.compact_packs() - new_pack = repository.chunks[w_id].pack_id - assert new_pack != pack_a # pack A was rewritten - new_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(new_pack)) + # the packs other than pack B: only the pack that replaced pack A. + packs = {i.name: i.size for i in repository.store_list("packs") if i.name != bin_to_hex(pack_b)} + ((new_pack_hex, new_size),) = packs.items() + assert new_pack_hex != bin_to_hex(pack_a) # pack A was rewritten assert new_size == pack_a_size - y_size # only the unused Y was dropped - reader = PackReader(store=repository.store, pack_id=new_pack) + reader = PackReader(store=repository.store, pack_id=hex_to_bin(new_pack_hex)) assert x_id in [chunk_id for chunk_id, _, _ in reader.iter_headers()] # the gap copy of X is kept diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 9b3eaed8ce..7fe9c09276 100644 --- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py +++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py @@ -4,8 +4,8 @@ import pytest from ...constants import * # NOQA -from ...helpers import bin_to_hex, sig_int, Error, CompressionSpec -from ...repository import Repository, PackReader, repo_lister +from ...helpers import bin_to_hex, hex_to_bin, sig_int, Error, CompressionSpec, IntegrityError +from ...repository import Repository, PackReader, PackTracker, repo_lister from ...cache import list_chunkindex_hashes from ...manifest import Manifest from ...compress import ZSTD, ZLIB, LZ4, CNONE @@ -349,9 +349,11 @@ def test_transform_pack_drops_superseded_gap(tmp_path): assert pdchunk(repository.get(H(1))) == b"XXXX" # the authoritative copy in pack B -def test_repo_compress_keeps_duplicate_indexed_in_missing_pack(archiver): +@pytest.mark.parametrize("untrusted", ("missing", "corrupt", "truncated")) +def test_repo_compress_keeps_duplicate_indexed_in_untrusted_pack(archiver, untrusted): # X is indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from the - # store, the copy in pack A is the only one left: repo-compress keeps it (#10474). + # store, recorded corrupt or truncated before the end of X, the copy in pack A may be the only + # readable one: repo-compress keeps it (#10474). cmd(archiver, "repo-create", RK_ENCRYPTION) with open_repository(archiver) as repository: repo_objs = Manifest.load(repository).repo_objs @@ -362,19 +364,39 @@ def test_repo_compress_keeps_duplicate_indexed_in_missing_pack(archiver): repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) repository.flush() pack_a = repository.chunks[w_id].pack_id - repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) - repository.flush() - pack_b = repository.chunks[x_id].pack_id - assert pack_b != pack_a - repository.store_delete("packs/" + bin_to_hex(pack_b)) - - cmd(archiver, "repo-compress", "-C", "none") + # repo-compress processes the packs in pack id order and stops with an error at a truncated + # pack: pack A must be rewritten before, so pack B needs the higher pack id. The pack id + # depends on the random nonce of X. + while True: + repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + pack_b = repository.chunks[x_id].pack_id + key_b = "packs/" + bin_to_hex(pack_b) + if pack_b > pack_a: + break + repository.store_delete(key_b) + if untrusted == "missing": + repository.store_delete(key_b) + elif untrusted == "corrupt": + tracker = PackTracker.load(repository) + tracker.record(pack_b, False) + tracker.save() + else: + repository.store_store(key_b, repository.store_load(key_b)[:-1]) + + if untrusted == "truncated": + with pytest.raises(IntegrityError, match="object extends past end of file"): + cmd(archiver, "repo-compress", "-C", "none") + else: + cmd(archiver, "repo-compress", "-C", "none", exit_code=EXIT_WARNING if untrusted == "corrupt" else 0) with open_repository(archiver) as repository: - new_pack = repository.chunks[w_id].pack_id - assert new_pack != pack_a # W was recompressed, pack A was rewritten - reader = PackReader(store=repository.store, pack_id=new_pack) - assert x_id in [chunk_id for chunk_id, _, _ in reader.iter_headers()] # the gap copy of X is kept + # the packs other than pack B: only the pack that replaced pack A. + (new_pack_hex,) = {info.name for info in repository.store_list("packs")} - {bin_to_hex(pack_b)} + assert new_pack_hex != bin_to_hex(pack_a) # W was recompressed, pack A was rewritten + reader = PackReader(store=repository.store, pack_id=hex_to_bin(new_pack_hex)) + # W and the gap copy of X are kept. + assert [chunk_id for chunk_id, _, _ in reader.iter_headers()] == [w_id, x_id] def test_transform_pack_unchanged_pack_untouched(tmp_path): From 8211bb7d764910b8735859cbf2d06b1f72225bfc Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sun, 4 Oct 2026 23:16:50 +0530 Subject: [PATCH 3/3] compact: reclaim superseded duplicates left after a full check --repair, fixes #10471 - pass 1 walks the gaps of the packs that have any and counts their superseded duplicates as reclaimable. A pack holding only unused indexed objects and superseded duplicates is deleted. The walk refreshes the lock per pack, shows progress and stops on Ctrl-C. - compact_pack drops the superseded duplicates found in pass 1 (superseded_ranges). - superseded_gap_ranges reads GAP_READ_SIZE bytes after a small object, so small objects share a store request, and META_READ_SIZE bytes after a large one. - compact and repo-compress check the index entries of every pack. A pack in which they overlap or reach past the end of the file is treated as untrusted. compact logs it as an error and keeps it unchanged. - a tiny pack that the all-packs gate keeps from being dropped or rewritten is a merge candidate. --- docs/internals/data-structures.rst | 7 +- docs/internals/packs.rst | 40 ++- src/borg/archiver/compact_cmd.py | 192 ++++++---- src/borg/archiver/repo_compress_cmd.py | 23 +- src/borg/repository.py | 60 +++- .../testsuite/archiver/compact_cmd_test.py | 332 +++++++++++++++++- .../archiver/repo_compress_cmd_test.py | 32 +- 7 files changed, 573 insertions(+), 113 deletions(-) diff --git a/docs/internals/data-structures.rst b/docs/internals/data-structures.rst index 6fc134aa88..4962aadb73 100644 --- a/docs/internals/data-structures.rst +++ b/docs/internals/data-structures.rst @@ -232,9 +232,10 @@ Compaction - free space one pack file at a time - a single object can not be removed from a pack file, only a whole pack file can be deleted or rewritten: - - a pack file whose indexed objects are all unused is deleted - - a pack file with some unused objects is rewritten without them, but only if - the wasted bytes reach the ``--threshold`` percentage + - a pack file holding only unused objects and superseded duplicates (copies of + chunks the chunks index records at another location) is deleted + - a pack file with some unused objects or superseded duplicates is rewritten + without them, but only if their 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 diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index d0581f9e50..df419ef0ce 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -280,18 +280,14 @@ stored as index fragment(s). A crash between steps 1 and 2 leaves blobs in ``packs/`` that no index entry covers (see `Gap bytes`_). No archive references these chunks. As the index does not list them, a later backup stores the chunks it needs again, which makes those blobs -superseded duplicates. ``borg compact`` reclaims unused indexed objects, and the -superseded duplicates in the packs it rewrites; a pack without index entries is not -rewritten, so its bytes stay (a tiny one can be merged as a whole). +superseded duplicates. ``borg compact`` reclaims unused indexed objects and superseded +duplicates; the other blobs no index entry covers stay. A full ``borg check --repair`` (``--repository-only`` rebuilds the index only if it is corrupt) rebuilds the index from the packs (see :ref:`pack-recovery`) and indexes one copy per chunk id, so the blobs whose chunk id had no index entry become indexed, and ``borg compact`` reclaims them once unused. The other copies stay superseded -duplicates and are only reclaimed when ``borg compact`` rewrites their pack to reclaim -unused indexed objects. -TODO: count superseded duplicates as reclaimable, so ``borg compact`` also rewrites a -pack that holds only used objects and superseded duplicates, see :issue:`10471`. +duplicates, which ``borg compact`` reclaims as well. A crash between steps 2 and 3 leaves index entries for objects no archive references. They point to valid, fully-written pack data, and ``borg compact`` @@ -302,8 +298,9 @@ archive pointer write is the commit point: archives are listed from the ``archives/`` namespace, so data not referenced by any archive pointer is unreachable, and ``borg compact`` treats its indexed objects as unused. -Pack files are removed by ``borg compact`` (dropping packs whose indexed objects are -all unused, rewriting packs above ``--threshold`` and merging tiny packs), +Pack files are removed by ``borg compact`` (dropping packs that hold only unused +objects and superseded gap blobs, see `Gap bytes`_, rewriting packs above +``--threshold`` and merging tiny packs), ``borg check --repair`` (when it drops a defective object, and when it salvages a pack recorded corrupt), ``borg repo-compress`` (``Repository.transform_pack`` stores the re-compressed pack under its new content-addressed name and deletes the old one) @@ -332,14 +329,27 @@ Validation covers ``meta_size`` and ``data_size``, so a dropped range is exactly Without a validator (``validate=None``), no gap bytes are dropped. A superseded blob is also kept when its indexed copy may be unreadable: ``borg compact`` and ``borg repo-compress`` keep it when that copy is in a pack that is missing from the store, recorded corrupt, or -shorter than its index entries state, ``borg check --repair --verify-data`` when that copy is -in a missing pack or in a pack holding a defect chunk. Merging packs (``merge_packs``) copies +whose index entries overlap or reach past the end of the pack file, +``borg check --repair --verify-data`` when that copy is in a missing pack or in a pack holding +a defect chunk. Merging packs (``merge_packs``) copies whole pack files, so it keeps all gap bytes. -The walk over a gap steps from header to header by the blob size each header states. It ends -at a header that does not parse or that reaches past the gap. The rest of that gap is kept, -and so is a superseded blob that does not validate; both are logged as a warning with the -pack id and the offset. +``borg compact`` walks the gaps of every pack not recorded corrupt and adds the superseded +gap blobs to the pack's unused indexed bytes. The sum decides whether the pack is rewritten +(``--threshold``), and a pack holding only unused indexed objects and superseded gap blobs is +deleted. A rewrite drops the superseded gap blobs that this walk found. ``borg compact`` +checks the index entries of every pack: a pack in which they overlap or reach past the end +of the pack file is logged as an error and left unchanged; its gaps are not walked, and a +gap blob whose indexed copy is in such a pack is kept. If the space to reclaim in the whole +repository is too small for ``borg compact`` to delete or rewrite packs, a tiny pack whose +indexed objects are all used is merged, superseded gap blobs included. + +The walk over a gap steps from header to header by the blob size each header states. A read +starts at a blob header. It is ``GAP_READ_SIZE`` (64 kiB) at the start of a gap and after a +blob smaller than that, so small blobs share a store request, and ``META_READ_SIZE`` (1 kiB) +after a larger blob. The walk ends at a header that does not parse or that reaches past the gap. The rest of +that gap is kept, and so is a superseded blob that does not validate; both are logged as a +warning with the pack id and the offset. .. _pack-index-namespace: diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index ce44c19493..3246dca65e 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -1,3 +1,4 @@ +from array import array from collections import defaultdict from pathlib import Path @@ -15,7 +16,8 @@ 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, PackTracker +from ..repository import Repository, PackReader, PackTracker +from ..repository import check_pack_objects, decode_ranges, superseded_gap_ranges from ..logger import create_logger @@ -252,35 +254,82 @@ def mark_soft_deleted_used(self): except (Archive.DoesNotExist, Repository.ObjectNotFound, IntegrityError) as e: logger.warning(f"Soft-deleted archive {name} {hex_id} cannot be fully preserved: {e}") + def superseded_duplicates(self, gap_ranges, pack_total, validate, untrusted_pack_ids): + """Return pack_id -> the ranges of the superseded duplicates in that pack, for the packs holding any. + + gap_ranges: pack_id -> the offset-ordered ranges of the pack's index entries, for the packs to walk. + pack_total: pack_id -> file size in the store. + validate, untrusted_pack_ids: passed to superseded_gap_ranges, which defines gaps and superseded + duplicates. + + The ranges are arrays of encoded byte ranges, see decode_ranges. + + Ctrl-C stops the walk at a pack boundary. + """ + superseded = {} + pi = ProgressIndicatorPercent( + total=len(gap_ranges), msg="Reading unindexed pack data %3.0f%%", msgid="compact.superseded_duplicates" + ) + for pid, ranges in gap_ranges.items(): + if sig_int: + break + pi.show() + # PackReader reads from the store, which does not refresh the repository lock. + self.repository._lock_refresh() + reader = PackReader(store=self.repository.store, pack_id=pid) + drop_ranges = superseded_gap_ranges( + reader, + self.chunks, + pid, + decode_ranges(ranges), + pack_total[pid], + validate=validate, + untrusted_pack_ids=untrusted_pack_ids, + ) + if drop_ranges: + superseded[pid] = array("Q", (offset << 32 | size for offset, size in drop_ranges)) + pi.finish() + return superseded + def compact_packs(self): """Free space one pack at a time (the store can only delete whole packs). - analyze_archives() has flagged the used objects F_USED. Only indexed-but-unused bytes are - reclaimed; bytes no index entry covers are preserved (see below). Per pack: + analyze_archives() has flagged the used objects F_USED. A pack's reclaimable bytes are its + indexed-but-unused bytes plus its superseded duplicates (see superseded_gap_ranges); other bytes + no index entry covers are preserved (see below). Per pack: - - all indexed objects unused, whole file indexed -> delete the pack. - - some indexed objects unused -> rewrite if the unused bytes reach - --threshold percent: copy the used objects (and any unindexed bytes) - into a new pack via compact_pack and drop the old one. Below the + - no indexed objects used, all bytes reclaimable -> delete the pack. + - some bytes reclaimable -> rewrite if the reclaimable bytes reach + --threshold percent: copy the used objects (and the other unindexed + bytes) into a new pack via compact_pack and drop the old one. Below the threshold keep the pack, so we don't rewrite a large pack to reclaim little. - - no indexed objects unused -> keep it, unless it is a tiny pack we + - no bytes reclaimable -> keep it, unless it is a tiny pack we merge (see below). + A tiny pack without unused indexed objects is also a merge candidate when its superseded + duplicates make it a delete or rewrite candidate, but the all-packs gate cancels that. + + A pack in which the index lists overlapping objects or an object past the end of the pack is + logged as an error and kept unchanged. + 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. + from a backup that crashed before writing its index). Except for the superseded duplicates, + compact keeps those; recovering or dropping them is "borg check --repair"'s job. See issue #9868. Compacting anything rewrites the whole chunk index and invalidates every client's cached copy, so compact only when the reclaimable space or the combined size of tiny packs is worth that cost. - Two passes bound the memory use: the first keeps only per-pack byte counts to pick the packs - to change, the second collects object ids for just those packs, not the whole index. + Two passes bound the memory use: the first keeps per-pack byte counts and the byte range of + each index entry as one int (see decode_ranges) to pick the packs to change, the second collects + object ids for just those packs, not the whole index. The first pass also walks the gaps of the + packs that have any (superseded_duplicates). The rewrites drop the superseded duplicates that + walk found. A pack's size is the file size the store reports, so it also counts bytes no index entry covers. @@ -298,9 +347,9 @@ def compact_packs(self): # detect which packs hold unindexed bytes. collect index entries whose pack is not in the store. pack_used = defaultdict(int) pack_indexed = defaultdict(int) # pack_id -> bytes of all its index entries, used or not + pack_ranges = {pid: array("Q") for pid in pack_total} # pack_id -> byte ranges of all its index entries stale_ids = [] # index entries referencing a pack file that is not in the store - stale_packs = set() # ids of those missing pack files - short_packs = set() # ids of the pack files that end before one of their index entries does + stale_packs = set() # ids of the pack files that are not in the store stale_used = 0 # how many of those were still flagged used (lost data) for id, entry in self.chunks.iteritems(): pid = entry.pack_id @@ -311,8 +360,7 @@ def compact_packs(self): stale_used += 1 else: pack_indexed[pid] += entry.obj_size - if entry.obj_offset + entry.obj_size > pack_total[pid]: - short_packs.add(pid) + pack_ranges[pid].append(entry.obj_offset << 32 | entry.obj_size) if entry.flags & ChunkIndex.F_USED: pack_used[pid] += entry.obj_size @@ -326,58 +374,83 @@ def compact_packs(self): logger.error(f"{stale_used} of them are still in use: repository data is missing!") set_ec(EXIT_ERROR) + # packs recorded corrupt in PackTracker that are still in the store + corrupt_packs = set(PackTracker.load(self.repository).corrupt_ids()) & pack_total.keys() + + # check the index entries of each pack against its file size. + inconsistent_packs = set() # ids of the packs whose index entries overlap or reach past the end of the file + gap_ranges = {} # pack_id -> offset-ordered byte ranges of its index entries, for the packs to walk + for pid, total in pack_total.items(): + ranges = array("Q", sorted(pack_ranges.pop(pid))) + try: + check_pack_objects(bin_to_hex(pid), decode_ranges(ranges), total) + except IntegrityError as err: + logger.error(str(err)) + set_ec(EXIT_ERROR) + inconsistent_packs.add(pid) + else: + # corrupt packs are not rewritten or merged, so their gaps are not walked. + if total > pack_indexed[pid] and pid not in corrupt_packs: + gap_ranges[pid] = ranges + + validate = object_validator(self.manifest.repo_objs) + untrusted_pack_ids = stale_packs | corrupt_packs | inconsistent_packs + # pack_id -> byte ranges of its superseded duplicates + pack_superseded = self.superseded_duplicates(gap_ranges, pack_total, validate, untrusted_pack_ids) + del gap_ranges + if sig_int: + return repo_size_before, repo_size_before # Ctrl-C during the walk, nothing was changed + superseded_bytes = {pid: sum(r & 0xFFFFFFFF for r in ranges) for pid, ranges in pack_superseded.items()} + # unindexed bytes: pack bytes no index entry covers, e.g. the objects of an interrupted borg create. - # superseded duplicates are unindexed objects whose chunk id the index maps to another location. - # compact_pack drops those when it rewrites a pack and keeps the other unindexed objects, as + # compact reclaims the superseded duplicates among them and keeps the other unindexed objects, as # "borg check --repair" can recover them (#9868). A full check --repair (--repository-only rebuilds # the index only if it is corrupt) indexes one copy per chunk id, so objects whose chunk id had no - # index entry become reclaimable once unused, and the other copies stay superseded duplicates. - # TODO(#10471): count superseded duplicates as reclaimable, so compact rewrites a pack that holds only - # used objects and superseded duplicates, e.g. after a crashed borg create was re-run. + # index entry become reclaimable once unused, and the other copies become superseded duplicates. unindexed = sum(total - pack_indexed[pid] for pid, total in pack_total.items() if total > pack_indexed[pid]) + unindexed -= sum(superseded_bytes.values()) if unindexed: logger.info( f"{format_file_size(unindexed)} in pack files is not covered by the index. " '"borg check --repair" (without --repository-only) indexes the objects whose chunk id has no ' - 'index entry, so "borg compact" reclaims them once unused. Copies of chunks indexed elsewhere ' - "are only reclaimed when compact rewrites their pack." + 'index entry, so "borg compact" reclaims them once unused.' ) - # 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. + # decide each pack's fate. # a merge fills packs up to pack_max_size, so cap "tiny" at half of that: a merged full pack # is then at least twice tiny_limit and no longer a merge candidate. tiny_limit = min(MIN_PACK_SIZE, self.repository.pack_max_size // 2) drop_packs, rewrite_packs, merge_packs = set(), set(), set() + tiny_reclaim_packs = set() # tiny drop or rewrite candidates without unused indexed bytes pack_reclaim = {} # pack_id -> reclaimable bytes, for the packs we act on (drop or rewrite) for pid, total in pack_total.items(): indexed = pack_indexed[pid] used = pack_used[pid] - if indexed > total: - # the index lists more bytes than the file holds. - logger.error(f'Pack {bin_to_hex(pid)}: index claims more data than the file holds, run "borg check".') - set_ec(EXIT_ERROR) + if pid in inconsistent_packs: 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 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 + unused = indexed - used # unused indexed bytes + superseded = superseded_bytes.get(pid, 0) + reclaimable = unused + superseded # the only bytes compact removes + if reclaimable and used == 0 and indexed + superseded == total: + drop_packs.add(pid) # whole file is unused indexed bytes and superseded duplicates -> drop it pack_reclaim[pid] = reclaimable - 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 + elif reclaimable and 100 * reclaimable / total >= self.threshold and pid not in corrupt_packs: + rewrite_packs.add(pid) # wasteful enough -> copy used objects (and other unindexed bytes) forward pack_reclaim[pid] = reclaimable + elif unused == 0 and total < tiny_limit and pid not in corrupt_packs: + merge_packs.add(pid) # fully-used but tiny -> merge candidate # else: below threshold -> leave alone + if pid in pack_reclaim and unused == 0 and total < tiny_limit and pid not in corrupt_packs: + tiny_reclaim_packs.add(pid) # all-packs gate: drop/rewrite only when the space they free reaches threshold/5 percent of # all pack bytes; freeing less does not justify the whole-index rewrite. reclaimable_total = sum(pack_reclaim.values()) worth_reclaiming = repo_size_before > 0 and 100 * reclaimable_total / repo_size_before >= self.threshold / 5 + if not worth_reclaiming: + # the tiny packs that are not dropped or rewritten for their superseded duplicates are merge candidates. + drop_packs, rewrite_packs, pack_reclaim = set(), set(), {} + merge_packs |= tiny_reclaim_packs # merge only once the tiny packs' combined size reaches a full pack, so every merge produces # at least one pack too large to ever be a merge candidate again. merging smaller amounts # would just repack a still-tiny pile into a slightly-bigger-but-still-tiny pack, and each @@ -385,8 +458,6 @@ def compact_packs(self): tiny_total = sum(pack_total[pid] for pid in merge_packs) worth_merging = tiny_total >= self.repository.pack_max_size - if not worth_reclaiming: - drop_packs, rewrite_packs, pack_reclaim = set(), set(), {} if not worth_merging: merge_packs = set() @@ -409,7 +480,7 @@ def compact_packs(self): return repo_size_before, repo_size_before # dry run: report only, change nothing if not drop_packs and not rewrite_packs and not merge_packs: - logger.info("Deleting 0 unused objects...") + logger.info(f"Deleting 0 unused objects and 0 superseded duplicates, freeing {format_file_size(0)}...") return repo_size_before, repo_size_before # nothing worth doing; chunk indexes stay valid # crash-safety (#9748): invalidate chunk indexes before the first store change. This also @@ -432,10 +503,15 @@ def compact_packs(self): forget[pid].add(id) # deleted counts index objects removed: every object of a dropped pack, plus the unused - # objects cut from rewritten packs. reclaimed counts the bytes freed. + # objects cut from rewritten packs. superseded counts the superseded duplicates in the dropped + # and rewritten packs. reclaimed counts the bytes freed. deleted = sum(len(ids) for ids in forget.values()) + sum(len(ids) for ids in drop.values()) + superseded = sum(len(pack_superseded.get(pid, ())) for pid in pack_reclaim) reclaimed = sum(pack_reclaim.values()) - logger.info(f"Deleting {deleted} unused objects, freeing {format_file_size(reclaimed)}...") + logger.info( + f"Deleting {deleted} unused objects and {superseded} superseded duplicates, " + f"freeing {format_file_size(reclaimed)}..." + ) pi = ProgressIndicatorPercent( total=len(drop_packs) + len(rewrite_packs) + (1 if merge_packs else 0), msg="Compacting packs %3.1f%%", @@ -459,8 +535,6 @@ def compact_packs(self): del self.chunks[id] progress += 1 pi.show(progress) # report after the work, so the final pack lands on 100% - validate = object_validator(self.manifest.repo_objs) - untrusted_pack_ids = stale_packs | corrupt_packs | short_packs for pid in rewrite_packs: if sig_int: break @@ -472,7 +546,7 @@ def compact_packs(self): drop_ids=drop[pid], chunks=self.chunks, validate=validate, - untrusted_pack_ids=untrusted_pack_ids, + superseded_ranges=decode_ranges(pack_superseded.get(pid, ())), ) freed += dropped # unused indexed objects plus superseded duplicates progress += 1 @@ -517,14 +591,13 @@ def build_parser_compact(self, subparsers, common_parser, mid_common_parser): - backups of source files that encountered an I/O error mid-transfer and were skipped - corruption of the repository (e.g., the archives directory lost entries; see notes below) - ``borg compact`` reclaims objects the chunk index knows about, plus redundant copies of - indexed chunks (e.g. written by concurrent backups) that it finds while rewriting a pack. - Other bytes no index entry covers, such as packs left behind by a backup that crashed - before recording its objects, are kept. ``borg check --repair`` (without - ``--repository-only``) indexes one copy per chunk id, so ``borg compact`` reclaims the - objects whose chunk id had no index entry once they are unused. The remaining copies of - chunks indexed elsewhere (e.g. when the crashed backup was re-run) stay unindexed and are - only reclaimed when ``borg compact`` rewrites their pack to reclaim unused indexed objects. + ``borg compact`` reclaims unused objects the chunk index knows about, plus copies of + chunks the index records at another location (e.g. written by concurrent backups, or by a + crashed backup that was re-run). Other bytes no index entry covers, such as packs left + behind by a backup that crashed before recording its objects, are kept. + ``borg check --repair`` (without ``--repository-only``) indexes one copy per chunk id, + so ``borg compact`` reclaims the objects whose chunk id had no index entry once they are + unused. ``borg compact`` does not rewrite or merge packs that ``borg check`` recorded as corrupt and warns about them. ``borg check --repair`` salvages such a pack: it replaces it by a pack @@ -582,6 +655,7 @@ def build_parser_compact(self, subparsers, common_parser, mid_common_parser): dest="threshold", type=int, default=10, - help="rewrite a pack when at least PERCENT of its bytes are unused; also gates whether " - "to compact at all (see the all-packs gate above), 0 disables that gate (default: 10)", + help="rewrite a pack when at least PERCENT of its bytes are unused objects or copies of chunks " + "indexed at another location; also gates whether to compact at all (see the all-packs gate above), " + "0 disables that gate (default: 10)", ) diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py index 449a1c2542..02c0859c1c 100644 --- a/src/borg/archiver/repo_compress_cmd.py +++ b/src/borg/archiver/repo_compress_cmd.py @@ -1,4 +1,5 @@ import logging +from array import array from collections import defaultdict from borgstore.store import ItemInfo @@ -8,10 +9,10 @@ from ..compress import ObfuscateSize, Auto, COMPRESSOR_TABLE from ..constants import * # NOQA 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 import format_file_size, bin_to_hex, hex_to_bin, IntegrityError from ..helpers.argparsing import ArgumentParser from ..repoobj import object_validator -from ..repository import Repository, PackTracker +from ..repository import Repository, PackTracker, check_pack_objects, decode_ranges from ..logger import create_logger @@ -90,12 +91,20 @@ def recompress(self): # group the indexed objects per pack; transform_pack requires each pack's complete id list. per_pack = defaultdict(list) # pack_id -> [chunk_id, ...] - short_packs = set() # ids of the pack files that end before one of their index entries does + pack_ranges = {pack_id: array("Q") for pack_id in pack_sizes} # pack_id -> byte ranges of its index entries for id, entry in self.chunks.iteritems(): per_pack[entry.pack_id].append(id) - pack_size = pack_sizes.get(entry.pack_id) - if pack_size is not None and entry.obj_offset + entry.obj_size > pack_size: - short_packs.add(entry.pack_id) + ranges = pack_ranges.get(entry.pack_id) + if ranges is not None: + ranges.append(entry.obj_offset << 32 | entry.obj_size) + + # transform_pack raises IntegrityError for these packs. + inconsistent_packs = set() # ids of the packs whose index entries overlap or reach past the end of the file + for pack_id, pack_size in packs: + try: + check_pack_objects(bin_to_hex(pack_id), decode_ranges(sorted(pack_ranges.pop(pack_id))), pack_size) + except IntegrityError: + inconsistent_packs.add(pack_id) self.packs_count = len(packs) size_before = sum(size for _, size in packs) @@ -126,7 +135,7 @@ def recompress(self): total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress" ) validate = object_validator(self.repo_objs) - untrusted_pack_ids = stale_packs | corrupt_packs | short_packs + untrusted_pack_ids = stale_packs | corrupt_packs | inconsistent_packs for i, (pack_id, pack_size) in enumerate(packs): if sig_int: break # stop cleanly at a pack boundary: save the index below, then raise diff --git a/src/borg/repository.py b/src/borg/repository.py index 730fd25a2d..e3c19c40dc 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -46,6 +46,9 @@ # how much to read to get an object's header plus, at the usual metadata slot sizes, its metadata # slot in the same read. META_READ_SIZE = 1024 +# how much of a gap superseded_gap_ranges reads at once after an object smaller than this, so that +# small objects share a read. after a larger object, it reads META_READ_SIZE. +GAP_READ_SIZE = 64 * 1024 # the largest metadata slot a validating read fetches. a slot holds a few compression fields, packed # and encrypted, i.e. some tens of bytes, so this is a generous bound and it keeps a corrupt # meta_size, which only MAX_DATA_SIZE bounds, from triggering a large read. @@ -575,6 +578,16 @@ def iter_headers(self, validate=None, on_drop=None): offset += obj_size +def decode_ranges(ranges): + """Return the (offset, size) tuples of byte ranges encoded as offset << 32 | size. + + borg compact and borg repo-compress keep the byte ranges of all index entries, so they store each + as one int in an array. Offset and size are below 2**32, like obj_offset and obj_size in the + chunks index. + """ + return [(r >> 32, r & 0xFFFFFFFF) for r in ranges] + + def check_pack_objects(pack_hex, obj_ranges, pack_size): """Validate a pack's indexed objects against the pack's file size. @@ -604,10 +617,13 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val the stored size of the indexed copy. Each gap is walked from object header to object header, stepping by the object size the header - states. An object is reported when it is a superseded duplicate and validate accepts it. The - walk over a gap ends at a header that does not parse or that reaches past the gap. Objects - validate rejects are kept, and so is the rest of a gap where the walk ends early; both are - logged as a warning with the pack id and the offset. + states. A read starts at an object header. It is GAP_READ_SIZE bytes at the start of a gap and + after an object smaller than that, so small objects share a read, and META_READ_SIZE bytes after + a larger object. + An object is reported when it is a superseded duplicate and validate accepts it. The walk over + a gap ends at a header that does not parse or that reaches past the gap. Objects validate + rejects are kept, and so is the rest of a gap where the walk ends early; both are logged as a + warning with the pack id and the offset. reader: PackReader of the pack. chunks: the chunks index (chunk id -> ChunkIndexEntry). @@ -640,11 +656,15 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val drop_ranges = [] # (obj_offset, obj_size) of superseded duplicates, offset-ordered hdr_size = RepoObj.obj_header.size for gstart, gend in gaps: + buf, buf_offset = memoryview(b""), gstart # buf holds the gap bytes from buf_offset on + read_size = GAP_READ_SIZE offset = gstart while offset < gend: - # one read for the header and the metadata slot after it. - buf = reader.read(offset, min(gend - offset, META_READ_SIZE)) - hdr, problem = PackReader._parse_header(buf[:hdr_size], offset, pack_size) + pos = offset - buf_offset # position of the object in buf + if pos + META_READ_SIZE > len(buf) and buf_offset + len(buf) < gend: + # buf does not hold the header and the usual metadata slot of this object: read from here on. + buf, buf_offset, pos = memoryview(reader.read(offset, min(gend - offset, read_size))), offset, 0 + hdr, problem = PackReader._parse_header(buf[pos : pos + hdr_size], offset, pack_size) if hdr is not None and offset + hdr_size + hdr.meta_size + hdr.data_size > gend: hdr, problem = None, "object reaching past its gap" if hdr is None: @@ -660,7 +680,7 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val and entry.pack_id not in untrusted_pack_ids and (entry.pack_id != pack_id or entry.obj_offset != offset) ): - problem = reader._validation_problem(hdr, offset, buf, offset, validate) + problem = reader._validation_problem(hdr, offset, buf, buf_offset, validate) if problem is None: drop_ranges.append((offset, obj_size)) else: @@ -672,6 +692,7 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, val # be a copy of an object inside another object's unencrypted data, whose range can cover # bytes of the gap objects after it, which would then be dropped. offset += obj_size + read_size = GAP_READ_SIZE if obj_size < GAP_READ_SIZE else META_READ_SIZE return drop_ranges @@ -2160,6 +2181,7 @@ def compact_pack( chunks=None, before_old_pack_delete=None, untrusted_pack_ids=frozenset(), + superseded_ranges=None, ): """Rewrite pack , keeping and dropping , then delete the old pack. @@ -2171,6 +2193,9 @@ def compact_pack( before_old_pack_delete: callable without arguments, called once just before the old pack is deleted. Not called when no bytes are dropped, since the old pack then stays. untrusted_pack_ids: passed to superseded_gap_ranges. + superseded_ranges: the (offset, size) ranges of the superseded duplicates to drop, as superseded_gap_ranges + returned them for this pack and this index. Default: superseded_gap_ranges is called here. If given, + validate and untrusted_pack_ids are not used. Together, keep_ids and drop_ids must cover every object the chunk index lists for this pack; an unlisted indexed object would keep its bytes in the new pack but its index entry would go @@ -2214,13 +2239,20 @@ def compact_pack( # record the dropped objects' byte ranges; every other byte (kept objects and gaps that no # index entry covers) is copied into the new pack unchanged. superseded duplicates found in # the gaps are dropped along with them (see superseded_gap_ranges). - # TODO(#9868 follow-up): classify gaps in compact_packs pass 1 too, so superseded bytes count - # toward the rewrite threshold and a wholly superseded orphan pack can be dropped outright. drop_ranges = [(offset, size) for offset, _, size, keep in located if not keep] - reader = PackReader(store=self.store, pack_id=pack_id) - drop_ranges += superseded_gap_ranges( - reader, chunks, pack_id, obj_ranges, pack_size, validate=validate, untrusted_pack_ids=untrusted_pack_ids - ) + if superseded_ranges is None: + reader = PackReader(store=self.store, pack_id=pack_id) + superseded_ranges = superseded_gap_ranges( + reader, chunks, pack_id, obj_ranges, pack_size, validate=validate, untrusted_pack_ids=untrusted_pack_ids + ) + else: + # superseded duplicates lie in gaps: they overlap neither the listed objects nor each other. + covered = 0 + for offset, size in sorted(obj_ranges + superseded_ranges): + assert offset >= covered, "superseded_ranges overlap an indexed object or each other" + covered = offset + size + assert covered <= pack_size, "superseded_ranges reach past the end of the pack" + drop_ranges += superseded_ranges drop_ranges.sort() dropped_bytes = sum(size for _, size in drop_ranges) # on-disk bytes this rewrite frees, for --stats diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index e1f7f26ccf..b0d7766fdf 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -6,10 +6,10 @@ import pytest from ...constants import * # NOQA -from ...helpers import get_cache_dir, bin_to_hex, hex_to_bin, sig_int, Error +from ...helpers import get_cache_dir, bin_to_hex, hex_to_bin, sig_int, Error, format_file_size, get_reset_ec from ...hashindex import ChunkIndex from ...repoobj import RepoObj -from ...repository import Repository, PackReader, PackTracker +from ...repository import Repository, PackReader, PackTracker, GAP_READ_SIZE, META_READ_SIZE 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 @@ -360,6 +360,334 @@ def test_compact_keeps_duplicate_indexed_in_untrusted_pack(tmp_path, untrusted): assert x_id in [chunk_id for chunk_id, _, _ in reader.iter_headers()] # the gap copy of X is kept +def put_pack(repository, repo_objs, datas): + """Store datas as repo objects in one new pack, return (chunk ids, pack id).""" + ids = [repo_objs.id_hash(data) for data in datas] + repository._pack_writer.max_count = len(datas) # one flush() -> one pack + for cid, data in zip(ids, datas): + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) + repository.flush() + return ids, repository.chunks[ids[0]].pack_id + + +def pack_sizes(repository): + return {hex_to_bin(info.name): info.size for info in repository.store_list("packs")} + + +def mark_used(repository, ids): + for cid, entry in repository.chunks.iteritems(): + flags = ChunkIndex.F_USED if cid in ids else ChunkIndex.F_NONE + repository.chunks[cid] = entry._replace(flags=flags) + + +def test_compact_drops_pack_of_superseded_duplicates(tmp_path, caplog): + # A crashed borg create left pack A, the re-run stored its chunks again in pack B (#10471). + # Pack A holds only superseded duplicates: compact deletes it. + caplog.set_level(logging.INFO) + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + _, pack_a = put_pack(repository, repo_objs, datas) + ids, pack_b = put_pack(repository, repo_objs, datas + [b"new"]) + mark_used(repository, set(ids)) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + assert "Deleting 0 unused objects and 2 superseded duplicates" in caplog.text + assert pack_sizes(repository).keys() == {pack_b} + for cid, data in zip(ids, datas + [b"new"]): + assert repo_objs.parse(cid, repository.get(cid), ro_type=ROBJ_FILE_STREAM)[1] == data + assert "not covered by the index" not in caplog.text + + +@pytest.mark.parametrize("threshold, rewritten", ((10, True), (90, False))) +def test_compact_rewrites_pack_for_superseded_duplicates(tmp_path, threshold, rewritten): + # Pack A holds the used W and superseded duplicates of X and Y, indexed in pack B. Their bytes + # count toward --threshold. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + w, x, y = os.urandom(1000), os.urandom(1000), os.urandom(1000) + (w_id, x_id, y_id), _ = put_pack(repository, repo_objs, [w, x, y]) + w_size = repository.chunks[w_id].obj_size + put_pack(repository, repo_objs, [x, y]) + mark_used(repository, {w_id, x_id, y_id}) + sizes_before = pack_sizes(repository) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=threshold) + gc.chunks = repository.chunks + gc.compact_packs() + + new_pack = repository.chunks[w_id].pack_id + if rewritten: + assert pack_sizes(repository)[new_pack] == w_size # only W was copied into the new pack + else: + assert pack_sizes(repository) == sizes_before + for cid, data in [(w_id, w), (x_id, x), (y_id, y)]: + assert repo_objs.parse(cid, repository.get(cid), ro_type=ROBJ_FILE_STREAM)[1] == data + + +def test_compact_dry_run_counts_superseded_duplicates(tmp_path, caplog): + caplog.set_level(logging.INFO) + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + _, pack_a = put_pack(repository, repo_objs, datas) + ids, _ = put_pack(repository, repo_objs, datas + [b"new"]) + mark_used(repository, set(ids)) + sizes_before = pack_sizes(repository) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10, dry_run=True) + gc.chunks = repository.chunks + gc.compact_packs() + + assert f"Would free {format_file_size(sizes_before[pack_a])} by dropping 1 packs" in caplog.text + assert pack_sizes(repository) == sizes_before + + +def test_compact_keeps_pack_of_duplicates_indexed_in_missing_pack(tmp_path): + # Pack A holds only copies of chunks indexed in pack B, which is missing from the store: these + # copies are not superseded duplicates, compact keeps pack A (#10474). + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + _, pack_a = put_pack(repository, repo_objs, datas) + ids, pack_b = put_pack(repository, repo_objs, datas + [b"new"]) + repository.store_delete("packs/" + bin_to_hex(pack_b)) + mark_used(repository, set(ids)) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + assert pack_a in pack_sizes(repository) + + +def test_compact_keeps_duplicates_indexed_in_truncated_pack(tmp_path): + # Pack A holds only copies of X and Y, which are indexed in pack B. Pack B is truncated before X: + # compact keeps pack A. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + _, pack_a = put_pack(repository, repo_objs, datas) + ids, pack_b = put_pack(repository, repo_objs, [b"new"] + datas) + key_b = "packs/" + bin_to_hex(pack_b) + repository.store_store(key_b, repository.store_load(key_b)[:100]) + mark_used(repository, set(ids)) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + assert pack_a in pack_sizes(repository) + + +@pytest.mark.parametrize("case", ("overlap", "overlap_in_gap_pack", "oversized_entry")) +def test_compact_keeps_duplicates_indexed_in_inconsistent_pack(tmp_path, caplog, case): + # Pack A holds only copies of X and Y, which are indexed in pack B. The index entries of pack B + # overlap. Compact logs the error for pack B, keeps packs A and B and drops pack C, which holds only + # an unused object. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + _, pack_a = put_pack(repository, repo_objs, datas) + (new_id, x_id, y_id), pack_b = put_pack(repository, repo_objs, [b"new"] + datas) + put_pack(repository, repo_objs, [os.urandom(1000)]) + chunks = repository.chunks + if case == "overlap": # Y at the offset of X: the entries of pack B still sum to its file size + chunks[y_id] = chunks[y_id]._replace(obj_offset=chunks[x_id].obj_offset) + elif case == "overlap_in_gap_pack": + del chunks[new_id] + chunks[y_id] = chunks[y_id]._replace(obj_offset=chunks[x_id].obj_offset + 1) + else: # the first entry covers the whole pack: the entries of pack B sum to more than its file size + chunks[new_id] = chunks[new_id]._replace(obj_size=pack_sizes(repository)[pack_b]) + mark_used(repository, {new_id, x_id, y_id}) + + get_reset_ec() + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + assert get_reset_ec() == EXIT_ERROR + assert f"pack {bin_to_hex(pack_b)}: overlapping objects" in caplog.text + assert pack_sizes(repository).keys() == {pack_a, pack_b} + + +@pytest.mark.parametrize("count, size", ((20, 100), (3, 100000))) +def test_compact_gap_walk_store_requests(tmp_path, monkeypatch, count, size): + # Pack A has no index entries, so it is one gap. Objects smaller than GAP_READ_SIZE share a store + # request. After an object larger than that, the walk reads META_READ_SIZE bytes. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(size) for _ in range(count)] + _, pack_a = put_pack(repository, repo_objs, datas) + ids, pack_b = put_pack(repository, repo_objs, datas + [b"new"]) + assert pack_b != pack_a + mark_used(repository, set(ids)) + pack_a_size = pack_sizes(repository)[pack_a] + + read_sizes = [] # the size of each read of pack A + store_load = repository.store.load + + def load(name, **kw): + if name == "packs/" + bin_to_hex(pack_a): + read_sizes.append(kw["size"]) + return store_load(name, **kw) + + monkeypatch.setattr(repository.store, "load", load) + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10, dry_run=True) + gc.chunks = repository.chunks + gc.compact_packs() + + if size < GAP_READ_SIZE: + assert pack_a_size < GAP_READ_SIZE + assert read_sizes == [pack_a_size] + else: + assert read_sizes == [GAP_READ_SIZE] + [META_READ_SIZE] * (count - 1) + + +@pytest.mark.parametrize("ctrl_c", (False, True)) +def test_compact_gap_walk_refreshes_lock_and_stops_on_ctrl_c(tmp_path, monkeypatch, ctrl_c): + # Packs A1 and A2 hold only superseded duplicates. The lock is refreshed before each pack is walked. + # Ctrl-C during the walk of the first pack stops the walk, and compact changes nothing. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + datas = [os.urandom(1000), os.urandom(1000)] + put_pack(repository, repo_objs, datas[:1]) + put_pack(repository, repo_objs, datas[1:]) + ids, pack_b = put_pack(repository, repo_objs, datas + [b"new"]) + mark_used(repository, set(ids)) + sizes_before = pack_sizes(repository) + + events = [] + lock_refresh, gap_ranges = repository._lock_refresh, compact_cmd.superseded_gap_ranges + + def walk(*args, **kw): + events.append("walk") + sig_int._sig_int_triggered = ctrl_c + return gap_ranges(*args, **kw) + + monkeypatch.setattr(repository, "_lock_refresh", lambda: events.append("refresh") or lock_refresh()) + monkeypatch.setattr(compact_cmd, "superseded_gap_ranges", walk) + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + try: + gc.compact_packs() + finally: + sig_int._sig_int_triggered = False # reset the global flag for the following tests + + walks = [i for i, event in enumerate(events) if event == "walk"] + if ctrl_c: + assert len(walks) == 1 + assert gc.store_changed is False + assert pack_sizes(repository) == sizes_before + else: + first, second = walks + assert "refresh" in events[first + 1 : second] # the walk itself does not refresh the lock + assert pack_sizes(repository).keys() == {pack_b} + + +@pytest.mark.parametrize("pack_b_fate", ("dropped", "rewritten")) +def test_compact_drops_duplicate_of_unused_chunk(tmp_path, caplog, pack_b_fate): + # Pack A holds the used W and a copy of X. X is unused and indexed in pack B, which compact drops + # (it holds nothing else) or rewrites (it also holds the used V). Compact removes X from pack B and + # rewrites pack A without its copy of X, as logged. + caplog.set_level(logging.INFO) + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + w, x, v = os.urandom(1000), os.urandom(1000), os.urandom(1000) + (w_id, x_id), pack_a = put_pack(repository, repo_objs, [w, x]) + used = {w_id} + if pack_b_fate == "dropped": + put_pack(repository, repo_objs, [x]) + else: + (_, v_id), _ = put_pack(repository, repo_objs, [x, v]) + used.add(v_id) + kept = sum(repository.chunks[cid].obj_size for cid in used) + mark_used(repository, used) + freed = sum(pack_sizes(repository).values()) - kept + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + size_before, size_after = gc.compact_packs() + + assert ( + f"Deleting 1 unused objects and 1 superseded duplicates, freeing {format_file_size(freed)}" in caplog.text + ) + assert size_before - size_after == freed + assert x_id not in gc.chunks + assert pack_sizes(repository) == {gc.chunks[cid].pack_id: gc.chunks[cid].obj_size for cid in used} + + +def test_compact_nothing_to_do_message(tmp_path, caplog): + caplog.set_level(logging.INFO) + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + ids, _ = put_pack(repository, manifest.repo_objs, [os.urandom(1000)]) + mark_used(repository, set(ids)) + + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + + assert "Deleting 0 unused objects and 0 superseded duplicates, freeing 0 B..." in caplog.text + assert gc.store_changed is False + + +def test_compact_merges_tiny_pack_not_rewritten_for_superseded_duplicates(tmp_path, monkeypatch): + # The tiny pack A holds the used W and a superseded duplicate of X, which reaches --threshold. The + # space to reclaim in all packs is below the all-packs gate, so pack A is not rewritten. It is + # merged with the other tiny packs. + monkeypatch.setenv("BORG_PACK_MAX_SIZE", "4000") + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + w, x = b"W", os.urandom(200) + (w_id, x_id), pack_a = put_pack(repository, repo_objs, [w, x]) + put_pack(repository, repo_objs, [x]) + objects = {w_id: w, x_id: x} + while sum(pack_sizes(repository).values()) < repository.pack_max_size: + data = os.urandom(50) + (cid,), _ = put_pack(repository, repo_objs, [data]) + objects[cid] = data + mark_used(repository, set(objects)) + packs_before = pack_sizes(repository).keys() + + merged = [] # the pack id sets passed to merge_packs + merge_packs = repository.merge_packs + monkeypatch.setattr( + repository, "merge_packs", lambda ids, **kw: merged.append(set(ids)) or merge_packs(ids, **kw) + ) + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=50) + gc.chunks = repository.chunks + gc.compact_packs() + + assert merged == [set(packs_before)] # all packs are merge candidates, pack A included + for cid, data in objects.items(): + assert repo_objs.parse(cid, repository.get(cid), ro_type=ROBJ_FILE_STREAM)[1] == data + + def test_compact_keeps_orphan_pack(tmp_path): # A pack with no index entries at all (a backup that crashed after writing the pack but before # writing its index) is left alone: its objects hold no reclaimable indexed waste and may be live diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 7fe9c09276..a3481b9215 100644 --- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py +++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py @@ -6,7 +6,7 @@ from ...constants import * # NOQA from ...helpers import bin_to_hex, hex_to_bin, sig_int, Error, CompressionSpec, IntegrityError from ...repository import Repository, PackReader, PackTracker, repo_lister -from ...cache import list_chunkindex_hashes +from ...cache import list_chunkindex_hashes, write_chunkindex_to_repo from ...manifest import Manifest from ...compress import ZSTD, ZLIB, LZ4, CNONE from ...archiver.repo_compress_cmd import PackRecompressor @@ -349,26 +349,27 @@ def test_transform_pack_drops_superseded_gap(tmp_path): assert pdchunk(repository.get(H(1))) == b"XXXX" # the authoritative copy in pack B -@pytest.mark.parametrize("untrusted", ("missing", "corrupt", "truncated")) +@pytest.mark.parametrize("untrusted", ("missing", "corrupt", "truncated", "overlap")) def test_repo_compress_keeps_duplicate_indexed_in_untrusted_pack(archiver, untrusted): - # X is indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from the - # store, recorded corrupt or truncated before the end of X, the copy in pack A may be the only - # readable one: repo-compress keeps it (#10474). + # X and Z are indexed in pack B, pack A holds another copy of X in a gap. With pack B missing from + # the store, recorded corrupt, truncated or with overlapping index entries, the copy in pack A may + # be the only readable one: repo-compress keeps it (#10474). cmd(archiver, "repo-create", RK_ENCRYPTION) with open_repository(archiver) as repository: repo_objs = Manifest.load(repository).repo_objs - w, x = b"W" * 1000, b"X" * 1000 - w_id, x_id = repo_objs.id_hash(w), repo_objs.id_hash(x) + w, x, z = b"W" * 1000, b"X" * 1000, b"Z" * 1000 + w_id, x_id, z_id = repo_objs.id_hash(w), repo_objs.id_hash(x), repo_objs.id_hash(z) repository._pack_writer.max_count = 2 # one flush() -> one pack for cid, data in [(w_id, w), (x_id, x)]: repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) repository.flush() pack_a = repository.chunks[w_id].pack_id # repo-compress processes the packs in pack id order and stops with an error at a truncated - # pack: pack A must be rewritten before, so pack B needs the higher pack id. The pack id - # depends on the random nonce of X. + # pack or at one with overlapping index entries: pack A must be rewritten before, so pack B + # needs the higher pack id. The pack id depends on the random nonces of X and Z. while True: - repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) + for cid, data in [(x_id, x), (z_id, z)]: + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) repository.flush() pack_b = repository.chunks[x_id].pack_id key_b = "packs/" + bin_to_hex(pack_b) @@ -381,11 +382,16 @@ def test_repo_compress_keeps_duplicate_indexed_in_untrusted_pack(archiver, untru tracker = PackTracker.load(repository) tracker.record(pack_b, False) tracker.save() - else: + elif untrusted == "truncated": repository.store_store(key_b, repository.store_load(key_b)[:-1]) + else: + chunks = repository.chunks + chunks[z_id] = chunks[z_id]._replace(obj_offset=chunks[x_id].obj_offset + 1) + write_chunkindex_to_repo(repository, chunks, incremental=False, force_write=True, delete_other=True) - if untrusted == "truncated": - with pytest.raises(IntegrityError, match="object extends past end of file"): + if untrusted in ("truncated", "overlap"): + error = "object extends past end of file" if untrusted == "truncated" else "overlapping objects" + with pytest.raises(IntegrityError, match=error): cmd(archiver, "repo-compress", "-C", "none") else: cmd(archiver, "repo-compress", "-C", "none", exit_code=EXIT_WARNING if untrusted == "corrupt" else 0)