From 10562c529d97d62c174a353327f26cba237f7341 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 10 Jul 2026 15:11:47 +0530 Subject: [PATCH 1/7] compact: add all-packs reclaim gate and tiny-pack merging Any compaction rewrites the whole chunk index, invalidating every client's cached copy, so only compact when the reclaimable space crosses a repo-wide threshold. Also merge small, fully-used packs once enough of them accumulate, since incremental backups otherwise leave one tiny pack behind per run with no way to consolidate them. --- src/borg/archiver/compact_cmd.py | 46 +++++++++--- src/borg/constants.py | 5 ++ src/borg/repository.py | 64 +++++++++++++++++ .../testsuite/archiver/compact_cmd_test.py | 72 ++++++++++++++++++- 4 files changed, 177 insertions(+), 10 deletions(-) diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 6de8a484b6..21ccaf3c25 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -35,6 +35,7 @@ def __init__(self, repository, manifest, *, stats, iec, threshold, dry_run=False self.threshold = threshold # rewrite a mixed pack only when its wasted-bytes fraction reaches this percent self.iec = iec # formats statistics using IEC units (1KiB = 1024B) self.dry_run = dry_run + self.store_changed = False # True once compact_packs() has changed the store (and its index) def garbage_collect(self): """Removes unused chunks from a repository.""" @@ -44,7 +45,8 @@ def garbage_collect(self): (self.missing_chunks, self.total_files, self.total_size, self.archives_count) = self.analyze_archives() self.report_and_delete() if not self.dry_run: - self.save_chunk_index() + if self.store_changed: + self.save_chunk_index() if sig_int: # raise after saving, so a Ctrl-C still leaves a valid index raise Error("Got Ctrl-C / SIGINT.") self.cleanup_files_cache() @@ -245,18 +247,23 @@ def compact_packs(self): 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 the pack. + - no indexed objects unused -> keep it, unless it is a tiny pack we + merge (see below). 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. + Compacting anything rewrites the whole chunk index and invalidates every client's cached + copy, so compact only when the reclaimable space or the tiny-pack count 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. A pack's size is the file size the store reports, so it also counts bytes no index entry covers. Returns (repo_size_before, repo_size_after), the on-disk pack size before and after this run. + Also sets self.store_changed True if the store (and thus the chunk index) was modified. """ # Pass 1: list the pack files and their sizes; these are the packs we consider. pack_total = {} # pack_id -> file size in the store @@ -301,7 +308,7 @@ def compact_packs(self): # 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. - drop_packs, rewrite_packs = set(), set() + drop_packs, rewrite_packs, merge_packs = set(), set(), set() 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] @@ -313,6 +320,8 @@ def compact_packs(self): continue # leave this pack untouched reclaimable = indexed - used # unused indexed bytes; the only bytes compact removes if reclaimable == 0: + if used == indexed and total < MIN_PACK_SIZE: + 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 @@ -321,19 +330,35 @@ def compact_packs(self): rewrite_packs.add(pid) # wasteful enough -> copy used objects (and unindexed bytes) forward pack_reclaim[pid] = reclaimable # else: below threshold -> leave alone + + # 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 + # merge only from MIN_PACK_COUNT tiny packs up, so merging does not run for every backup's + # single new pack. + worth_merging = len(merge_packs) >= MIN_PACK_COUNT + + if not worth_reclaiming: + drop_packs, rewrite_packs, pack_reclaim = set(), set(), {} + if not worth_merging: + merge_packs = set() + if self.dry_run: freed = sum(pack_reclaim.values()) logger.info( - f"Would free {format_file_size(freed, iec=self.iec)} " - f"by dropping {len(drop_packs)} packs and rewriting {len(rewrite_packs)} packs." + f"Would free {format_file_size(freed, iec=self.iec)} by dropping {len(drop_packs)} " + f"packs, rewriting {len(rewrite_packs)} packs and merging {len(merge_packs)} tiny packs." ) return repo_size_before, repo_size_before # dry run: report only, change nothing - if not drop_packs and not rewrite_packs: + + if not drop_packs and not rewrite_packs and not merge_packs: logger.info("Deleting 0 unused objects...") - return repo_size_before, repo_size_before # nothing to reclaim; chunk indexes stay valid + 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 delete_chunkindex_from_repo(self.repository) + self.store_changed = True # Pass 2: collect object ids only for the affected packs (a subset, not the whole index) keep = defaultdict(set) # rewrite pack_id -> its used objects, kept in the new pack @@ -352,7 +377,7 @@ def compact_packs(self): reclaimed = sum(pack_reclaim.values()) logger.info(f"Deleting {deleted} unused objects, freeing {format_file_size(reclaimed, iec=self.iec)}...") pi = ProgressIndicatorPercent( - total=len(drop_packs) + len(rewrite_packs), + total=len(drop_packs) + len(rewrite_packs) + (1 if merge_packs else 0), msg="Compacting packs %3.1f%%", step=0.1, msgid="compact.compact_packs", @@ -383,6 +408,11 @@ def compact_packs(self): freed += dropped # unused indexed objects plus superseded duplicates progress += 1 pi.show(progress) + if merge_packs and not sig_int: + # merge the tiny packs into fewer, larger ones + self.repository.merge_packs(merge_packs, chunks=self.chunks) + progress += 1 + pi.show(progress) pi.finish() return repo_size_before, repo_size_before - freed diff --git a/src/borg/constants.py b/src/borg/constants.py index 336f626947..9801ba0e57 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -62,6 +62,11 @@ # default pack size limit [bytes], see PackWriter in the repository module DEFAULT_PACK_MAX_SIZE = 50 * 1000 * 1000 +# borg compact merges tiny packs once at least MIN_PACK_COUNT of them, +# each smaller than MIN_PACK_SIZE bytes, have accumulated. +MIN_PACK_SIZE = DEFAULT_PACK_MAX_SIZE // 50 # 1 MB +MIN_PACK_COUNT = 10 + # MAX_OBJECT_SIZE = MAX_DATA_SIZE + len(PUT header) MAX_OBJECT_SIZE = MAX_DATA_SIZE + 41 # see assertion at end of repository module diff --git a/src/borg/repository.py b/src/borg/repository.py index 9e0e403377..5bad2edf2b 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -1051,6 +1051,70 @@ def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None): self.store_delete(pack_key) return new_pack_id, dropped_bytes + def merge_packs(self, pack_ids, *, chunks=None, max_size=None): + """Combine several small packs into fewer, larger ones to reduce the pack count. + + pack_ids: the packs to merge. Each still-indexed object of these packs is copied into a new + pack (each kept at most max_size bytes); afterwards the source packs are deleted. + chunks: the ChunkIndex to read object locations from and to apply the index updates to. + Must be the index pack_ids were derived from. Default: self.chunks. + max_size: byte cap for each merged pack. Default: the repository's configured pack size limit. + + Every new pack is stored and indexed before any source pack is deleted, so a crash mid-merge + never destroys the only stored copy of an object. + """ + self._lock_refresh() + if chunks is None: + chunks = self.chunks + if max_size is None: + max_size = self._pack_writer.max_size or DEFAULT_PACK_MAX_SIZE + pack_ids = set(pack_ids) + + # collect every still-indexed object of the selected packs, grouped per source pack and + # ordered by offset so each source pack is read sequentially. + per_pack = {} # pack_id -> [(obj_offset, obj_id, obj_size), ...] + for obj_id, entry in chunks.iteritems(): + if entry.pack_id in pack_ids: + per_pack.setdefault(entry.pack_id, []).append((entry.obj_offset, obj_id, entry.obj_size)) + for objs in per_pack.values(): + objs.sort() + + # greedily batch objects across packs so each output pack stays within max_size. + batches = [] # each batch: [(src_pack_id, obj_offset, obj_id, obj_size), ...] + current, current_size = [], 0 + for pid, objs in per_pack.items(): + for offset, obj_id, size in objs: + if current and current_size + size > max_size: + batches.append(current) + current, current_size = [], 0 + current.append((pid, offset, obj_id, size)) + current_size += size + if current: + batches.append(current) + + # write each batch as a new pack (named sha256 of its content) and repoint its objects. + produced = set() + for batch in batches: + sources = [(bin_to_hex(pid), offset, size) for pid, offset, _, size in batch] + new_pack_id = hex_to_bin(self.store.defrag(sources, algorithm="sha256", namespace="packs")) + produced.add(new_pack_id) + new_locations = [] + new_offset = 0 + for pid, _, obj_id, size in batch: + new_locations.append((obj_id, new_pack_id, new_offset, size)) + new_offset += size + chunks.update_pack_info(new_locations) + + # delete the source packs now that every new pack is stored and indexed. skip any pack whose + # id a batch reproduced (identical bytes hash to the same name): it now holds merged data. + for pid in pack_ids: + if pid in produced: + continue + try: + self.store_delete("packs/" + bin_to_hex(pid)) + except StoreObjectNotFound: + logger.warning(f"Pack {bin_to_hex(pid)} to merge was already gone.") + def break_lock(self): Lock(self.store).break_lock() diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index 43ef5cc57d..56eac9c947 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -11,6 +11,7 @@ from ...cache import delete_chunkindex_from_repo, write_chunkindex_to_repo from ...manifest import Manifest from ...archive import Archive +from ...archiver.compact_cmd import ArchiveGarbageCollector from . import cmd, create_regular_file, create_src_archive, generate_archiver_tests, open_repository, RK_ENCRYPTION from . import changedir from ..repository_test import H, fchunk, pdchunk @@ -110,7 +111,6 @@ def test_compact_interrupted_does_not_poison_chunk_index(archivers, request, mon is conservative: the next client rebuilds the index from actual repository contents and re-uploads any deleted data. This is tested for both compact paths (default and --stats). """ - from ...archiver.compact_cmd import ArchiveGarbageCollector archiver = request.getfixturevalue(archivers) @@ -208,7 +208,6 @@ def test_compact_packs_respects_threshold(tmp_path): # wastes 2/3 of its bytes is rewritten down to its single kept object (and its old file deleted); the # pack that wastes only 1/3 stays untouched, since copying its kept objects to reclaim that little is # not worth it. This covers the rewrite, leave-alone and keep/drop split unique to multi-object packs. - from ...archiver.compact_cmd import ArchiveGarbageCollector location = os.fspath(tmp_path / "repo") with Repository(location, exclusive=True, create=True) as repository: @@ -469,6 +468,75 @@ def test_compact_skips_oversized_index_entry(tmp_path): assert pdchunk(repository.get(H(0))) == b"DATA" +def test_compact_packs_merges_tiny_packs(tmp_path): + # Incremental backups leave many tiny, fully-used packs behind (issue #9816): the current + # unused-bytes policy never touches them, so they pile up. Once at least MIN_PACK_COUNT of them + # accumulate, compact merges them into fewer, larger packs while keeping every object readable. + + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + num = MIN_PACK_COUNT + 2 # enough tiny packs to cross the merge threshold + for i in range(num): + repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i))) + repository.flush() # flush after each put -> one object per pack -> many tiny packs + + # mark every object used, so no pack qualifies for drop/rewrite; they only qualify for merging + for i in range(num): + entry = repository.chunks[H(i)] + repository.chunks[H(i)] = entry._replace(flags=ChunkIndex.F_USED) + + packs_before = {info.name for info in repository.store_list("packs")} + assert len(packs_before) == num # one tiny pack per object + + gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + assert gc.store_changed is True # the merge changed the store + + # all tiny objects fit under the pack size limit, so they collapse into a single merged pack + packs_after = {info.name for info in repository.store_list("packs")} + assert len(packs_after) == 1 + assert packs_after.isdisjoint(packs_before) # a brand-new pack, all originals deleted + # every object still reads back correctly from wherever it now lives + for i in range(num): + assert pdchunk(repository.get(H(i))) == f"DATA{i}".encode() + # the (rebuilt) chunk index only references packs that still exist + for id, entry in repository.chunks.iteritems(): + assert bin_to_hex(entry.pack_id) in packs_after + + +def test_compact_packs_below_all_packs_gate_changes_nothing(tmp_path): + # A tiny unused pack alongside a large used one: reclaiming it would free far less than the + # all-packs threshold (threshold/5, i.e. 2% at the default). Any compaction forces a full + # chunk-index rewrite that invalidates every client's cached index (#9817), so below the gate + # compact must leave the repo untouched rather than pay that cost for so little. + + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + # one big, fully-used pack (well above MIN_PACK_SIZE, so it is not a merge candidate) ... + repository.put(H(0), fchunk(b"U" * 2_000_000, chunk_id=H(0))) + repository.flush() + # ... and one tiny pack we would otherwise drop entirely (all its bytes unused) + repository.put(H(1), fchunk(b"x", chunk_id=H(1))) + repository.flush() + + repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED) + repository.chunks[H(1)] = repository.chunks[H(1)]._replace(flags=ChunkIndex.F_NONE) + + packs_before = {info.name for info in repository.store_list("packs")} + assert len(packs_before) == 2 + + gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + assert gc.store_changed is False # below the all-packs gate: nothing was touched + + # both packs (including the fully-unused tiny one) still exist, and both objects read back + assert {info.name for info in repository.store_list("packs")} == packs_before + assert pdchunk(repository.get(H(0))) == b"U" * 2_000_000 + assert pdchunk(repository.get(H(1))) == b"x" + + def test_compact_gc_after_index_loss(archivers, request): """When no chunk index exists (e.g. after an interrupted compact, #9748), compact rebuilds the index from the packs. The rebuilt entries must start unused (init_flags=F_NONE), From 1b34a16ffefe688634d3c01be697d6f66a77b6a1 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 10 Jul 2026 19:41:36 +0530 Subject: [PATCH 2/7] compact: refuse up front on a repo without write+delete permissions The all-packs reclaim gate can make compact_packs() a no-op when there is nothing worth reclaiming, so the chunk index rewrite (and the store write that used to trip borgstore's permission check) never happens. A read-only repo then silently succeeds instead of being refused. Add Repository.assert_writable(), a borg-level check of the namespace permission dict computed in __init__, and call it once up front in do_compact() before any analysis work, independent of what the gate decides. --- src/borg/archiver/compact_cmd.py | 4 ++++ src/borg/repository.py | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 21ccaf3c25..37f553bb1e 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -422,6 +422,10 @@ class CompactMixIn: @with_repository(exclusive=True, compatibility=(Manifest.Operation.DELETE,)) def do_compact(self, args, repository, manifest): """Collects garbage in the repository.""" + if not args.dry_run: + # refuse up front on a repo opened read-only, before the reclaim gate can decide there + # is nothing to do and make compaction a silent no-op. + repository.assert_writable() ArchiveGarbageCollector( repository, manifest, stats=args.stats, iec=args.iec, threshold=args.threshold, dry_run=args.dry_run ).garbage_collect() diff --git a/src/borg/repository.py b/src/borg/repository.py index 5bad2edf2b..52baafb839 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -10,6 +10,7 @@ from borgstore.backends.errors import BackendError as StoreBackendError from borgstore.backends.errors import BackendDoesNotExist as StoreBackendDoesNotExist from borgstore.backends.errors import BackendAlreadyExists as StoreBackendAlreadyExists +from borgstore.backends.errors import PermissionDenied as StorePermissionDenied from .constants import * # NOQA from .hashindex import ChunkIndex @@ -388,6 +389,9 @@ def __init__( self.store = Store(url, config=ns_config, permissions=permissions, cache_url=cache_url) except StoreBackendError as e: raise Error(str(e)) + # None means "all" (no restrictions); for rest:// the backend enforces permissions + # server-side, so the client does not check them (see above). + self.permissions = None if location.proto == "rest" else permissions self.store_opened = False self.version = None # long-running repository methods which emit log or progress output are responsible for calling @@ -1153,6 +1157,22 @@ def store_delete(self, name, *, deleted=False): self._lock_refresh() return self.store.delete(name, deleted=deleted) + def assert_writable(self): + """Raise PermissionDenied if the repo permissions forbid compaction. + + Compaction stores new packs and index fragments and deletes the old ones, so it needs + write (w/W) and delete (D) access to the packs/ and index/ namespaces. self.permissions + is None when no restrictions apply (BORG_REPO_PERMISSIONS=all). + """ + if self.permissions is None: + return + for namespace in ("packs", "index"): + granted = set(self.permissions.get(namespace, self.permissions.get("", ""))) + if not (granted & set("wW")) or "D" not in granted: + raise StorePermissionDenied( + f"Repository permissions do not allow compaction (need write and delete on {namespace}/)." + ) + def store_move(self, name, new_name=None, *, delete=False, undelete=False, deleted=False): self._lock_refresh() return self.store.move(name, new_name, delete=delete, undelete=undelete, deleted=deleted) From 18952570d7dc08f134f61d284e0d4118c74f2a3c Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 10 Jul 2026 20:01:19 +0530 Subject: [PATCH 3/7] compact: merge tiny packs by combined size, not count 10 tiny packs merging into a still-tiny pack would just get re-merged later, invalidating every client's index each time for no lasting gain. Trigger the merge only once the tiny packs total a full pack instead. --- src/borg/archiver/compact_cmd.py | 12 +++-- src/borg/constants.py | 6 +-- src/borg/repository.py | 7 ++- .../testsuite/archiver/compact_cmd_test.py | 46 ++++++++++++++++--- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 37f553bb1e..7e036308d7 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -255,7 +255,8 @@ def compact_packs(self): 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 tiny-pack count is worth that cost. + 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. @@ -335,9 +336,12 @@ def compact_packs(self): # 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 - # merge only from MIN_PACK_COUNT tiny packs up, so merging does not run for every backup's - # single new pack. - worth_merging = len(merge_packs) >= MIN_PACK_COUNT + # 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 + # such repack invalidates the chunk index for every client for no lasting gain. + 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(), {} diff --git a/src/borg/constants.py b/src/borg/constants.py index 9801ba0e57..eb100b8a99 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -62,10 +62,10 @@ # default pack size limit [bytes], see PackWriter in the repository module DEFAULT_PACK_MAX_SIZE = 50 * 1000 * 1000 -# borg compact merges tiny packs once at least MIN_PACK_COUNT of them, -# each smaller than MIN_PACK_SIZE bytes, have accumulated. +# borg compact merges packs smaller than MIN_PACK_SIZE bytes ("tiny") once their combined size +# reaches the repository's max pack size, so every merge produces at least one full-size pack that +# is never tiny again (avoids repeatedly re-merging a growing-but-still-tiny pack). MIN_PACK_SIZE = DEFAULT_PACK_MAX_SIZE // 50 # 1 MB -MIN_PACK_COUNT = 10 # MAX_OBJECT_SIZE = MAX_DATA_SIZE + len(PUT header) MAX_OBJECT_SIZE = MAX_DATA_SIZE + 41 # see assertion at end of repository module diff --git a/src/borg/repository.py b/src/borg/repository.py index 52baafb839..9fbbbee4cb 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -553,6 +553,11 @@ def open(self, *, exclusive, lock_wait=None, lock=True): self._pack_writer = PackWriter(self.store, repository=self, max_count=max_count, max_size=max_size) self.opened = True + @property + def pack_max_size(self): + """The configured byte cap for a pack (BORG_PACK_MAX_SIZE, or the default if count-bound).""" + return self._pack_writer.max_size or DEFAULT_PACK_MAX_SIZE + @property def chunks(self): """ChunkIndex mapping every known chunk id to its pack location. @@ -1071,7 +1076,7 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): if chunks is None: chunks = self.chunks if max_size is None: - max_size = self._pack_writer.max_size or DEFAULT_PACK_MAX_SIZE + max_size = self.pack_max_size pack_ids = set(pack_ids) # collect every still-indexed object of the selected packs, grouped per source pack and diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index 56eac9c947..d39b25a116 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -468,14 +468,16 @@ def test_compact_skips_oversized_index_entry(tmp_path): assert pdchunk(repository.get(H(0))) == b"DATA" -def test_compact_packs_merges_tiny_packs(tmp_path): +def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch): # Incremental backups leave many tiny, fully-used packs behind (issue #9816): the current - # unused-bytes policy never touches them, so they pile up. Once at least MIN_PACK_COUNT of them - # accumulate, compact merges them into fewer, larger packs while keeping every object readable. + # unused-bytes policy never touches them, so they pile up. Once their combined size reaches a + # full pack, compact merges them into fewer, larger packs while keeping every object readable. + # BORG_PACK_MAX_SIZE is set small here so a handful of tiny packs already cross that threshold. + monkeypatch.setenv("BORG_PACK_MAX_SIZE", "300") location = os.fspath(tmp_path / "repo") with Repository(location, exclusive=True, create=True) as repository: - num = MIN_PACK_COUNT + 2 # enough tiny packs to cross the merge threshold + num = 10 for i in range(num): repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i))) repository.flush() # flush after each put -> one object per pack -> many tiny packs @@ -487,16 +489,18 @@ def test_compact_packs_merges_tiny_packs(tmp_path): packs_before = {info.name for info in repository.store_list("packs")} assert len(packs_before) == num # one tiny pack per object + total_bytes = sum(repository.store.info("packs/" + name).size for name in packs_before) + assert total_bytes >= repository.pack_max_size # combined size crosses the merge threshold gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() assert gc.store_changed is True # the merge changed the store - # all tiny objects fit under the pack size limit, so they collapse into a single merged pack + # the tiny packs collapse into fewer, larger packs (each up to pack_max_size) packs_after = {info.name for info in repository.store_list("packs")} - assert len(packs_after) == 1 - assert packs_after.isdisjoint(packs_before) # a brand-new pack, all originals deleted + assert len(packs_after) < len(packs_before) + assert packs_after.isdisjoint(packs_before) # brand-new packs, all originals deleted # every object still reads back correctly from wherever it now lives for i in range(num): assert pdchunk(repository.get(H(i))) == f"DATA{i}".encode() @@ -505,6 +509,34 @@ def test_compact_packs_merges_tiny_packs(tmp_path): assert bin_to_hex(entry.pack_id) in packs_after +def test_compact_packs_below_merge_size_gate_leaves_tiny_packs(tmp_path, monkeypatch): + # Guards against the count-based trigger this replaced (#9816 review feedback): merging must not + # fire just because several tiny packs exist, only once their combined size could produce a full + # pack. Otherwise a small repack would invalidate the chunk index for every client for no lasting + # reduction in tiny-pack count (Thomas's "10 packs of 1 kB merge into a still-tiny 10 kB pack" + # scenario). + monkeypatch.setenv("BORG_PACK_MAX_SIZE", "100000") # far above what these tiny packs total + + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + for i in range(3): + repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i))) + repository.flush() + for i in range(3): + entry = repository.chunks[H(i)] + repository.chunks[H(i)] = entry._replace(flags=ChunkIndex.F_USED) + + packs_before = {info.name for info in repository.store_list("packs")} + assert len(packs_before) == 3 + + gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10) + gc.chunks = repository.chunks + gc.compact_packs() + assert gc.store_changed is False # combined tiny bytes stay far below one full pack: leave them alone + + assert {info.name for info in repository.store_list("packs")} == packs_before + + def test_compact_packs_below_all_packs_gate_changes_nothing(tmp_path): # A tiny unused pack alongside a large used one: reclaiming it would free far less than the # all-packs threshold (threshold/5, i.e. 2% at the default). Any compaction forces a full From 8cca880a173990c25c79ee40bf6d4f35172d0cba Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 10 Jul 2026 20:03:34 +0530 Subject: [PATCH 4/7] compact: merge whole pack files, fail safe on a corrupt index merge_packs batched individual indexed objects, so bytes no index entry covered (crashed writes, superseded duplicates) were silently dropped on every merge. Copy whole pack files instead, and raise before touching the store if an index entry overlaps another or extends past the pack's end. --- src/borg/repository.py | 83 +++++++++++++++++++------ src/borg/testsuite/repository_test.py | 88 +++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 20 deletions(-) diff --git a/src/borg/repository.py b/src/borg/repository.py index 9fbbbee4cb..d5935597d6 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -1,6 +1,7 @@ import os import sys import time +from collections import defaultdict from pathlib import Path from hashlib import sha256 @@ -1063,12 +1064,20 @@ def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None): def merge_packs(self, pack_ids, *, chunks=None, max_size=None): """Combine several small packs into fewer, larger ones to reduce the pack count. - pack_ids: the packs to merge. Each still-indexed object of these packs is copied into a new - pack (each kept at most max_size bytes); afterwards the source packs are deleted. + pack_ids: the packs to merge, whole files; afterwards the source packs are deleted. chunks: the ChunkIndex to read object locations from and to apply the index updates to. Must be the index pack_ids were derived from. Default: self.chunks. max_size: byte cap for each merged pack. Default: the repository's configured pack size limit. + Whole pack files are copied, not individual indexed objects, so bytes no index entry covers + (a chunk copy superseded by a later put, or objects from a backup that crashed before + writing its index) are carried forward into the merged pack rather than silently dropped. + + Before copying anything, every source pack's index entries are checked for overlap or for + claiming bytes past the pack's actual end -- either means a corrupt index, and copying such + a pack could silently truncate the corrupt object. Raises IntegrityError in that case, + leaving the store untouched; recovery is "borg check --repair"'s job. + Every new pack is stored and indexed before any source pack is deleted, so a crash mid-merge never destroys the only stored copy of an object. """ @@ -1080,38 +1089,72 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): pack_ids = set(pack_ids) # collect every still-indexed object of the selected packs, grouped per source pack and - # ordered by offset so each source pack is read sequentially. - per_pack = {} # pack_id -> [(obj_offset, obj_id, obj_size), ...] + # ordered by offset, used both to validate each pack and to repoint its objects afterwards. + per_pack = defaultdict(list) # pack_id -> [(obj_offset, obj_id, obj_size), ...] for obj_id, entry in chunks.iteritems(): if entry.pack_id in pack_ids: - per_pack.setdefault(entry.pack_id, []).append((entry.obj_offset, obj_id, entry.obj_size)) + per_pack[entry.pack_id].append((entry.obj_offset, obj_id, entry.obj_size)) for objs in per_pack.values(): objs.sort() - # greedily batch objects across packs so each output pack stays within max_size. - batches = [] # each batch: [(src_pack_id, obj_offset, obj_id, obj_size), ...] + # get each source pack's real file size, dropping any pack that is already gone (stale + # index entries, #9850) before validating or copying anything. store.info() reports a + # missing object via info.exists rather than raising, unlike store.delete(). + pack_size = {} + for pid in list(pack_ids): + info = self.store.info("packs/" + bin_to_hex(pid)) + if not info.exists: + logger.warning(f"Pack {bin_to_hex(pid)} to merge was already gone.") + pack_ids.discard(pid) + per_pack.pop(pid, None) + continue + pack_size[pid] = info.size + + # validate every remaining pack before writing anything (see docstring). + for pid in pack_ids: + covered = 0 + for offset, _, size in per_pack[pid]: + if offset < covered: + raise IntegrityError( + f"pack {bin_to_hex(pid)}: overlapping objects at offset {offset} " + f'(index corruption), run "borg check"' + ) + covered = offset + size + if covered > pack_size[pid]: + raise IntegrityError( + f"pack {bin_to_hex(pid)}: object extends past end of file at offset {covered} " + f'(index corruption), run "borg check"' + ) + + # greedily batch whole pack files so each output pack stays within max_size. batches never + # split a source pack, so bytes no index entry covers always travel with their pack instead + # of being dropped. + batches = [] # each batch: [pack_id, ...] current, current_size = [], 0 - for pid, objs in per_pack.items(): - for offset, obj_id, size in objs: - if current and current_size + size > max_size: - batches.append(current) - current, current_size = [], 0 - current.append((pid, offset, obj_id, size)) - current_size += size + for pid in pack_ids: + size = pack_size[pid] + if current and current_size + size > max_size: + batches.append(current) + current, current_size = [], 0 + current.append(pid) + current_size += size if current: batches.append(current) - # write each batch as a new pack (named sha256 of its content) and repoint its objects. + # write each batch as a new pack (named sha256 of its content) and repoint its objects: an + # object's new offset is the running byte total of the packs placed before its pack in the + # batch, plus its old offset within that pack. produced = set() for batch in batches: - sources = [(bin_to_hex(pid), offset, size) for pid, offset, _, size in batch] + sources = [(bin_to_hex(pid), 0, pack_size[pid]) for pid in batch] new_pack_id = hex_to_bin(self.store.defrag(sources, algorithm="sha256", namespace="packs")) produced.add(new_pack_id) new_locations = [] - new_offset = 0 - for pid, _, obj_id, size in batch: - new_locations.append((obj_id, new_pack_id, new_offset, size)) - new_offset += size + pack_base = 0 + for pid in batch: + for offset, obj_id, size in per_pack[pid]: + new_locations.append((obj_id, new_pack_id, pack_base + offset, size)) + pack_base += pack_size[pid] chunks.update_pack_info(new_locations) # delete the source packs now that every new pack is stored and indexed. skip any pack whose diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index b80212a4f1..3a46503508 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -552,6 +552,94 @@ def short_read(*args, **kwargs): assert H(1) in repository.chunks # still indexed: aborted before deleting the dropped id +def test_merge_packs_combines_whole_files(repo_fixtures, request): + # Several one-object packs merge into one, and every object reads back from its new location. + repository = get_repository_from_fixture(repo_fixtures, request) + with repository: + pack_ids = [] + for i in range(3): + repository.put(H(i), fchunk(f"DATA{i}".encode(), chunk_id=H(i))) + repository.flush() # one object per pack + pack_ids.append(repository.chunks[H(i)].pack_id) + + packs_before = {info.name for info in repository.store_list("packs")} + assert len(packs_before) == 3 + + repository.merge_packs(pack_ids) + + packs_after = {info.name for info in repository.store_list("packs")} + assert len(packs_after) == 1 + assert packs_after.isdisjoint(packs_before) # a brand-new pack, all originals deleted + for i in range(3): + assert pdchunk(repository.get(H(i))) == f"DATA{i}".encode() + assert bin_to_hex(repository.chunks[H(i)].pack_id) in packs_after + + +def test_merge_packs_carries_unindexed_bytes_forward(repo_fixtures, request): + # A pack byte range no index entry covers (e.g. a chunk copy superseded by a later put) must + # survive a merge unchanged, not be silently dropped: merge_packs copies whole pack files. + # Merged alongside a second pack, so the result is a genuinely new pack (a lone unchanged pack + # would defrag to identical bytes and thus the same content-addressed name). + chunk0 = fchunk(b"AAAA", chunk_id=H(0)) + chunk1 = fchunk(b"BBBB", chunk_id=H(1)) # this entry is removed from the index before merging + chunk2 = fchunk(b"CCCC", chunk_id=H(2)) # lives in a second, separate pack + repository = get_repository_from_fixture(repo_fixtures, request) + build_one_pack(repository, [(H(0), chunk0), (H(1), chunk1)]) + with repository: + repository.put(H(2), chunk2) + repository.flush() + pack1_id = repository.chunks[H(0)].pack_id + pack2_id = repository.chunks[H(2)].pack_id + pack1_size_before = repository.store.info("packs/" + bin_to_hex(pack1_id)).size + del repository.chunks[H(1)] # H(1)'s bytes are now unindexed, but still on disk + + repository.merge_packs({pack1_id, pack2_id}) + + new_pack_id = repository.chunks[H(0)].pack_id + assert new_pack_id == repository.chunks[H(2)].pack_id # both merged into the same new pack + new_pack_size = repository.store.info("packs/" + bin_to_hex(new_pack_id)).size + assert new_pack_size == pack1_size_before + len(chunk2) # H(1)'s bytes carried forward, not dropped + assert pdchunk(repository.get(H(0))) == b"AAAA" + assert pdchunk(repository.get(H(2))) == b"CCCC" + + +def test_merge_packs_detects_past_eof(repo_fixtures, request): + # An index entry claiming bytes past its pack's actual end means index corruption. merge_packs + # must raise before writing anything, so the only intact copy of the pack is never deleted. + chunk0 = fchunk(b"DATA0", chunk_id=H(0)) + repository = get_repository_from_fixture(repo_fixtures, request) + build_one_pack(repository, [(H(0), chunk0)]) + with repository: + pack_id = repository.chunks[H(0)].pack_id + entry = repository.chunks[H(0)] + repository.chunks[H(0)] = entry._replace(obj_size=entry.obj_size + 1) # claims 1 byte past EOF + + with pytest.raises(IntegrityError): + repository.merge_packs({pack_id}) + assert bin_to_hex(pack_id) in [info.name for info in repository.store_list("packs")] # untouched + + +def test_merge_packs_skips_pack_already_gone(repo_fixtures, request): + # A stale index entry pointing at a pack file the store no longer has (#9850) must not abort the + # merge: the pack is excluded with a warning and the rest of the batch still merges. + chunk0 = fchunk(b"DATA0", chunk_id=H(0)) + chunk1 = fchunk(b"DATA1", chunk_id=H(1)) + repository = get_repository_from_fixture(repo_fixtures, request) + with repository: + repository.put(H(0), chunk0) + repository.flush() + pack0_id = repository.chunks[H(0)].pack_id + repository.put(H(1), chunk1) + repository.flush() + pack1_id = repository.chunks[H(1)].pack_id + + repository.store_delete("packs/" + bin_to_hex(pack0_id)) # pack gone, index entry left stale + + repository.merge_packs({pack0_id, pack1_id}) + + assert pdchunk(repository.get(H(1))) == b"DATA1" # the still-present pack merged normally + + def test_list(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: for x in range(100): From 4928b231b120ceb8934302aa241bcbc11ac72c4f Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sat, 11 Jul 2026 14:09:42 +0530 Subject: [PATCH 5/7] compact: raise borg Error on restricted repo, refresh lock and wrap defrag in merge_packs --- src/borg/archiver/compact_cmd.py | 15 ++++++++++- src/borg/repository.py | 44 +++++++++++++++++--------------- 2 files changed, 37 insertions(+), 22 deletions(-) diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 7e036308d7..f1a0250b45 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -462,6 +462,18 @@ def build_parser_compact(self, subparsers, common_parser, mid_common_parser): either regularly (e.g., once a month, possibly together with ``borg check``) or when disk space needs to be freed. + Compacting anything rewrites the whole chunk index and invalidates every client's + cached copy of it, so ``borg compact`` only acts when the gain is worth that cost: + + - All-packs gate: it drops or rewrites packs only when the space they would free + reaches ``--threshold`` divided by 5 percent (2% at the default threshold) of the + total pack size. Below that floor it leaves the repository (and the chunk index) + untouched. Use ``--threshold 0`` to disable the gate and always compact. + - Tiny-pack merging: incremental backups tend to leave one small, fully-used pack per + run. ``borg compact`` combines such tiny packs into larger ones, but only once their + combined size is large enough to fill at least one full-size pack, so a merge always + produces a pack that will not be a merge candidate again. + **Important:** After compacting, it is no longer possible to use ``borg undelete`` to recover @@ -496,5 +508,6 @@ 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 (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)", ) diff --git a/src/borg/repository.py b/src/borg/repository.py index d5935597d6..7cf7354c47 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -11,7 +11,6 @@ from borgstore.backends.errors import BackendError as StoreBackendError from borgstore.backends.errors import BackendDoesNotExist as StoreBackendDoesNotExist from borgstore.backends.errors import BackendAlreadyExists as StoreBackendAlreadyExists -from borgstore.backends.errors import PermissionDenied as StorePermissionDenied from .constants import * # NOQA from .hashindex import ChunkIndex @@ -319,6 +318,11 @@ class PathPermissionDenied(Error): exit_mcode = 21 + class CompactionPermissionDenied(Error): + """Repository permissions do not allow compaction (need write and delete on {}/).""" + + exit_mcode = 24 + # Whole packs kept in memory for reads; the least recently used is evicted first. # Memory use is this count times the pack size. PACK_READER_CACHE_SIZE = 3 @@ -1071,12 +1075,11 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): Whole pack files are copied, not individual indexed objects, so bytes no index entry covers (a chunk copy superseded by a later put, or objects from a backup that crashed before - writing its index) are carried forward into the merged pack rather than silently dropped. + writing its index) are carried into the merged pack too. - Before copying anything, every source pack's index entries are checked for overlap or for - claiming bytes past the pack's actual end -- either means a corrupt index, and copying such - a pack could silently truncate the corrupt object. Raises IntegrityError in that case, - leaving the store untouched; recovery is "borg check --repair"'s job. + Each source pack's index entries are checked for overlap or for claiming bytes past the + pack's actual end before anything is written; either means a corrupt index. Raises + IntegrityError in that case and leaves the store untouched; repair is "borg check --repair". Every new pack is stored and indexed before any source pack is deleted, so a crash mid-merge never destroys the only stored copy of an object. @@ -1088,8 +1091,7 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): max_size = self.pack_max_size pack_ids = set(pack_ids) - # collect every still-indexed object of the selected packs, grouped per source pack and - # ordered by offset, used both to validate each pack and to repoint its objects afterwards. + # collect every still-indexed object of the selected packs, grouped per source pack, ordered by offset. per_pack = defaultdict(list) # pack_id -> [(obj_offset, obj_id, obj_size), ...] for obj_id, entry in chunks.iteritems(): if entry.pack_id in pack_ids: @@ -1097,9 +1099,8 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): for objs in per_pack.values(): objs.sort() - # get each source pack's real file size, dropping any pack that is already gone (stale - # index entries, #9850) before validating or copying anything. store.info() reports a - # missing object via info.exists rather than raising, unlike store.delete(). + # get each source pack's real file size; drop any pack already gone from the store (its + # index entry is stale). store.info() reports a missing object via info.exists, not by raising. pack_size = {} for pid in list(pack_ids): info = self.store.info("packs/" + bin_to_hex(pid)) @@ -1126,12 +1127,11 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): f'(index corruption), run "borg check"' ) - # greedily batch whole pack files so each output pack stays within max_size. batches never - # split a source pack, so bytes no index entry covers always travel with their pack instead - # of being dropped. + # greedily batch whole pack files so each output pack stays within max_size, in sorted id + # order so batch composition is reproducible. batches = [] # each batch: [pack_id, ...] current, current_size = [], 0 - for pid in pack_ids: + for pid in sorted(pack_ids): size = pack_size[pid] if current and current_size + size > max_size: batches.append(current) @@ -1146,8 +1146,12 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): # batch, plus its old offset within that pack. produced = set() for batch in batches: + self._lock_refresh() # refresh the lock per batch, the loop can run for a while sources = [(bin_to_hex(pid), 0, pack_size[pid]) for pid in batch] - new_pack_id = hex_to_bin(self.store.defrag(sources, algorithm="sha256", namespace="packs")) + try: + new_pack_id = hex_to_bin(self.store.defrag(sources, algorithm="sha256", namespace="packs")) + except ReadRangeError as e: # a source pack shrank or is corrupt + raise IntegrityError(f'merge_packs: {e}, run "borg check"') from e produced.add(new_pack_id) new_locations = [] pack_base = 0 @@ -1159,7 +1163,7 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): # delete the source packs now that every new pack is stored and indexed. skip any pack whose # id a batch reproduced (identical bytes hash to the same name): it now holds merged data. - for pid in pack_ids: + for pid in sorted(pack_ids): if pid in produced: continue try: @@ -1206,7 +1210,7 @@ def store_delete(self, name, *, deleted=False): return self.store.delete(name, deleted=deleted) def assert_writable(self): - """Raise PermissionDenied if the repo permissions forbid compaction. + """Raise CompactionPermissionDenied if the repo permissions forbid compaction. Compaction stores new packs and index fragments and deletes the old ones, so it needs write (w/W) and delete (D) access to the packs/ and index/ namespaces. self.permissions @@ -1217,9 +1221,7 @@ def assert_writable(self): for namespace in ("packs", "index"): granted = set(self.permissions.get(namespace, self.permissions.get("", ""))) if not (granted & set("wW")) or "D" not in granted: - raise StorePermissionDenied( - f"Repository permissions do not allow compaction (need write and delete on {namespace}/)." - ) + raise self.CompactionPermissionDenied(namespace) def store_move(self, name, new_name=None, *, delete=False, undelete=False, deleted=False): self._lock_refresh() From 1aa33c018dfb4446832b2c87ce7ddd02a69d79b6 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sat, 11 Jul 2026 18:34:34 +0530 Subject: [PATCH 6/7] compact: interleave pack merge with per-batch deletion, make it interruptible Cap the tiny-pack limit at half the max pack size so a merged full pack is never a merge candidate again. Refuse compaction up front on a restricted repo. --- src/borg/archiver/compact_cmd.py | 5 ++- src/borg/constants.py | 3 +- src/borg/repository.py | 37 +++++++++++-------- .../testsuite/archiver/compact_cmd_test.py | 10 ++++- .../archiver/restricted_permissions_test.py | 19 +++++++--- src/borg/testsuite/repository_test.py | 23 ++++++++++++ 6 files changed, 73 insertions(+), 24 deletions(-) diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index f1a0250b45..60fb0ee56f 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/src/borg/archiver/compact_cmd.py @@ -309,6 +309,9 @@ def compact_packs(self): # decide each pack's fate. a pack's reclaimable bytes are its indexed-but-unused bytes; the # redundant duplicates compact_pack finds in the gaps are reclaimed on top when it rewrites. + # a merge fills packs up to pack_max_size, so cap "tiny" at half of that: a merged full pack + # 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() pack_reclaim = {} # pack_id -> reclaimable bytes, for the packs we act on (drop or rewrite) for pid, total in pack_total.items(): @@ -321,7 +324,7 @@ def compact_packs(self): continue # leave this pack untouched reclaimable = indexed - used # unused indexed bytes; the only bytes compact removes if reclaimable == 0: - if used == indexed and total < MIN_PACK_SIZE: + if used == indexed and total < tiny_limit: merge_packs.add(pid) # fully-used but tiny -> merge candidate continue # nothing to reclaim -> leave alone if used == 0 and indexed == total: diff --git a/src/borg/constants.py b/src/borg/constants.py index eb100b8a99..27bfda0eb1 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -64,7 +64,8 @@ # borg compact merges packs smaller than MIN_PACK_SIZE bytes ("tiny") once their combined size # reaches the repository's max pack size, so every merge produces at least one full-size pack that -# is never tiny again (avoids repeatedly re-merging a growing-but-still-tiny pack). +# is never tiny again (avoids repeatedly re-merging a growing-but-still-tiny pack). compact caps the +# tiny limit at half the configured max pack size (see compact_packs). MIN_PACK_SIZE = DEFAULT_PACK_MAX_SIZE // 50 # 1 MB # MAX_OBJECT_SIZE = MAX_DATA_SIZE + len(PUT header) diff --git a/src/borg/repository.py b/src/borg/repository.py index 7cf7354c47..ee4b20501f 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -18,6 +18,7 @@ from .helpers import Location from .helpers import bin_to_hex, hex_to_bin from .helpers import get_cache_dir +from .helpers import sig_int from .helpers import ProgressIndicatorPercent from .helpers.lrucache import LRUCache from .storelocking import Lock @@ -1081,8 +1082,10 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): pack's actual end before anything is written; either means a corrupt index. Raises IntegrityError in that case and leaves the store untouched; repair is "borg check --repair". - Every new pack is stored and indexed before any source pack is deleted, so a crash mid-merge - never destroys the only stored copy of an object. + Packs are merged one batch at a time, each batch's sources deleted once its merged pack is + stored and indexed. So a crash or Ctrl-C between batches never destroys the only stored copy + of an object, and the store holds at most one batch of extra packs at a time. The packs not + yet merged are merged on the next run. """ self._lock_refresh() if chunks is None: @@ -1142,10 +1145,13 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): batches.append(current) # write each batch as a new pack (named sha256 of its content) and repoint its objects: an - # object's new offset is the running byte total of the packs placed before its pack in the - # batch, plus its old offset within that pack. - produced = set() + # object's new offset is the running byte total of the packs before its pack in the batch, + # plus its old offset within that pack. + pi = ProgressIndicatorPercent(total=len(batches), msg="Merging packs %3.0f%%", msgid="repository.merge_packs") + produced = set() # merged pack ids; a one-pack batch reproduces its source's id for batch in batches: + if sig_int: + break self._lock_refresh() # refresh the lock per batch, the loop can run for a while sources = [(bin_to_hex(pid), 0, pack_size[pid]) for pid in batch] try: @@ -1160,16 +1166,17 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None): new_locations.append((obj_id, new_pack_id, pack_base + offset, size)) pack_base += pack_size[pid] chunks.update_pack_info(new_locations) - - # delete the source packs now that every new pack is stored and indexed. skip any pack whose - # id a batch reproduced (identical bytes hash to the same name): it now holds merged data. - for pid in sorted(pack_ids): - if pid in produced: - continue - try: - self.store_delete("packs/" + bin_to_hex(pid)) - except StoreObjectNotFound: - logger.warning(f"Pack {bin_to_hex(pid)} to merge was already gone.") + # delete this batch's sources; skip any pack a batch reproduced (a one-pack batch hashes + # to the same content-addressed name), which now holds the merged data. + for pid in batch: + if pid in produced: + continue + try: + self.store_delete("packs/" + bin_to_hex(pid)) + except StoreObjectNotFound: + logger.warning(f"Pack {bin_to_hex(pid)} to merge was already gone.") + pi.show(increase=1) + pi.finish() def break_lock(self): Lock(self.store).break_lock() diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index d39b25a116..ef824cc580 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/src/borg/testsuite/archiver/compact_cmd_test.py @@ -500,7 +500,7 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch): # the tiny packs collapse into fewer, larger packs (each up to pack_max_size) packs_after = {info.name for info in repository.store_list("packs")} assert len(packs_after) < len(packs_before) - assert packs_after.isdisjoint(packs_before) # brand-new packs, all originals deleted + assert packs_after - packs_before # at least one genuinely new merged pack # every object still reads back correctly from wherever it now lives for i in range(num): assert pdchunk(repository.get(H(i))) == f"DATA{i}".encode() @@ -508,6 +508,14 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch): for id, entry in repository.chunks.iteritems(): assert bin_to_hex(entry.pack_id) in packs_after + # a merged full-size pack is no longer tiny (the tiny limit is pack_max_size // 2 here), so a + # second compact finds nothing to merge and leaves the store unchanged. + gc2 = ArchiveGarbageCollector(repository, manifest=None, stats=False, iec=False, threshold=10) + gc2.chunks = repository.chunks + gc2.compact_packs() + assert gc2.store_changed is False + assert {info.name for info in repository.store_list("packs")} == packs_after + def test_compact_packs_below_merge_size_gate_leaves_tiny_packs(tmp_path, monkeypatch): # Guards against the count-based trigger this replaced (#9816 review feedback): merging must not diff --git a/src/borg/testsuite/archiver/restricted_permissions_test.py b/src/borg/testsuite/archiver/restricted_permissions_test.py index 1b98f8c0db..926a4a56d5 100644 --- a/src/borg/testsuite/archiver/restricted_permissions_test.py +++ b/src/borg/testsuite/archiver/restricted_permissions_test.py @@ -4,6 +4,7 @@ from borgstore.backends.errors import PermissionDenied from ...constants import * # NOQA +from ...repository import Repository from .. import changedir from . import cmd, create_test_files, RK_ENCRYPTION, generate_archiver_tests @@ -75,10 +76,13 @@ def test_repository_permissions_no_delete(archivers, request, monkeypatch): # Verify the archive still exists. assert "archive2" in cmd(archiver, "repo-list") - # Try to compact the repo, which should fail. - with pytest.raises(PermissionDenied): + # Try to compact the repo, which should fail (no "D" on packs/): compact refuses up front. + with pytest.raises(Repository.CompactionPermissionDenied): cmd(archiver, "compact") + # A dry run only reads and reports, so it is allowed even without write/delete access. + cmd(archiver, "compact", "--dry-run") + # Check without --repair should work. cmd(archiver, "check") @@ -128,10 +132,13 @@ def test_repository_permissions_read_only(archivers, request, monkeypatch): with pytest.raises(PermissionDenied): cmd(archiver, "repo-delete") - # Try to compact the repo, which should fail. - with pytest.raises(PermissionDenied): + # Try to compact the repo, which should fail (no write/delete access): compact refuses up front. + with pytest.raises(Repository.CompactionPermissionDenied): cmd(archiver, "compact") + # A dry run only reads and reports, so it is allowed under read-only permissions. + cmd(archiver, "compact", "--dry-run") + def test_repository_permissions_write_only(archivers, request, monkeypatch): """Test repository with 'write-only' permissions setting""" @@ -170,8 +177,8 @@ def test_repository_permissions_write_only(archivers, request, monkeypatch): with pytest.raises(PermissionDenied): cmd(archiver, "delete", "archive1") - # Try to compact the repo, which should fail (data dir has "lw" permissions, no reading). - with pytest.raises(PermissionDenied): + # Try to compact the repo, which should fail (no "D" on packs/): compact refuses up front. + with pytest.raises(Repository.CompactionPermissionDenied): cmd(archiver, "compact") # Try to check the repo, which should fail (data dir has "lw" permissions, no reading). diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 3a46503508..c968c96bf6 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -640,6 +640,29 @@ def test_merge_packs_skips_pack_already_gone(repo_fixtures, request): assert pdchunk(repository.get(H(1))) == b"DATA1" # the still-present pack merged normally +def test_assert_writable(repository): + # Compaction needs write (w/W) and delete (D) on both packs/ and index/. assert_writable reads + # self.permissions directly, so set it explicitly to cover each case. + with repository: + repository.permissions = None # "all": no restrictions + repository.assert_writable() # does not raise + + repository.permissions = {"packs": "lrwWD", "index": "lrwWD"} + repository.assert_writable() # does not raise + + repository.permissions = {"packs": "lrw", "index": "lrwWD"} # packs/ has no delete + with pytest.raises(Repository.CompactionPermissionDenied): + repository.assert_writable() + + repository.permissions = {"packs": "lrwWD", "index": "lrD"} # index/ has no write + with pytest.raises(Repository.CompactionPermissionDenied): + repository.assert_writable() + + repository.permissions = {"": "lr"} # neither namespace listed; "" fallback grants read only + with pytest.raises(Repository.CompactionPermissionDenied): + repository.assert_writable() + + def test_list(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: for x in range(100): From 52d7ca0db1e5832a18a252f61b2ae7d994938624 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sat, 11 Jul 2026 20:12:00 +0530 Subject: [PATCH 7/7] compact: use a generic Repository.PermissionDenied with a detailed message --- src/borg/repository.py | 11 +++++++---- .../testsuite/archiver/restricted_permissions_test.py | 6 +++--- src/borg/testsuite/repository_test.py | 6 +++--- 3 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/borg/repository.py b/src/borg/repository.py index ee4b20501f..580507519f 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -319,8 +319,8 @@ class PathPermissionDenied(Error): exit_mcode = 21 - class CompactionPermissionDenied(Error): - """Repository permissions do not allow compaction (need write and delete on {}/).""" + class PermissionDenied(Error): + """Repository permission denied: {}""" exit_mcode = 24 @@ -1217,7 +1217,7 @@ def store_delete(self, name, *, deleted=False): return self.store.delete(name, deleted=deleted) def assert_writable(self): - """Raise CompactionPermissionDenied if the repo permissions forbid compaction. + """Raise PermissionDenied if the repo permissions forbid compaction. Compaction stores new packs and index fragments and deletes the old ones, so it needs write (w/W) and delete (D) access to the packs/ and index/ namespaces. self.permissions @@ -1228,7 +1228,10 @@ def assert_writable(self): for namespace in ("packs", "index"): granted = set(self.permissions.get(namespace, self.permissions.get("", ""))) if not (granted & set("wW")) or "D" not in granted: - raise self.CompactionPermissionDenied(namespace) + raise self.PermissionDenied( + f"compaction needs write (w/W) and delete (D) permissions on {namespace}/, " + f"but only {''.join(sorted(granted))!r} is granted (BORG_REPO_PERMISSIONS)." + ) def store_move(self, name, new_name=None, *, delete=False, undelete=False, deleted=False): self._lock_refresh() diff --git a/src/borg/testsuite/archiver/restricted_permissions_test.py b/src/borg/testsuite/archiver/restricted_permissions_test.py index 926a4a56d5..fbd9a8aff1 100644 --- a/src/borg/testsuite/archiver/restricted_permissions_test.py +++ b/src/borg/testsuite/archiver/restricted_permissions_test.py @@ -77,7 +77,7 @@ def test_repository_permissions_no_delete(archivers, request, monkeypatch): assert "archive2" in cmd(archiver, "repo-list") # Try to compact the repo, which should fail (no "D" on packs/): compact refuses up front. - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): cmd(archiver, "compact") # A dry run only reads and reports, so it is allowed even without write/delete access. @@ -133,7 +133,7 @@ def test_repository_permissions_read_only(archivers, request, monkeypatch): cmd(archiver, "repo-delete") # Try to compact the repo, which should fail (no write/delete access): compact refuses up front. - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): cmd(archiver, "compact") # A dry run only reads and reports, so it is allowed under read-only permissions. @@ -178,7 +178,7 @@ def test_repository_permissions_write_only(archivers, request, monkeypatch): cmd(archiver, "delete", "archive1") # Try to compact the repo, which should fail (no "D" on packs/): compact refuses up front. - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): cmd(archiver, "compact") # Try to check the repo, which should fail (data dir has "lw" permissions, no reading). diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index c968c96bf6..f9e4125979 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -651,15 +651,15 @@ def test_assert_writable(repository): repository.assert_writable() # does not raise repository.permissions = {"packs": "lrw", "index": "lrwWD"} # packs/ has no delete - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): repository.assert_writable() repository.permissions = {"packs": "lrwWD", "index": "lrD"} # index/ has no write - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): repository.assert_writable() repository.permissions = {"": "lr"} # neither namespace listed; "" fallback grants read only - with pytest.raises(Repository.CompactionPermissionDenied): + with pytest.raises(Repository.PermissionDenied): repository.assert_writable()