compact: add all-packs reclaim gate and tiny-pack merging

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,24 @@ 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 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.
 
         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 +309,10 @@ 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()
+        # 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():
             indexed = pack_indexed[pid]
@@ -313,6 +324,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 < tiny_limit:
+                    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 +334,38 @@ 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 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(), {}
+        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 +384,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 +415,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
@@ -392,6 +429,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()
@@ -424,6 +465,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
@@ -458,5 +511,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)",
         )

src/borg/constants.py

@@ -62,6 +62,12 @@
 # default pack size limit [bytes], see PackWriter in the repository module
 DEFAULT_PACK_MAX_SIZE = 50 * 1000 * 1000
 
+# 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). 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)
 MAX_OBJECT_SIZE = MAX_DATA_SIZE + 41  # see assertion at end of repository module
 

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
 
@@ -17,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
@@ -317,6 +319,11 @@ class PathPermissionDenied(Error):
 
         exit_mcode = 21
 
+    class PermissionDenied(Error):
+        """Repository permission denied: {}"""
+
+        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
@@ -388,6 +395,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
@@ -549,6 +559,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.
@@ -1051,6 +1066,118 @@ 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, 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 into the merged pack too.
+
+        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".
+
+        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:
+            chunks = self.chunks
+        if max_size is 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, 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:
+                per_pack[entry.pack_id].append((entry.obj_offset, obj_id, entry.obj_size))
+        for objs in per_pack.values():
+            objs.sort()
+
+        # 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))
+            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, in sorted id
+        # order so batch composition is reproducible.
+        batches = []  # each batch: [pack_id, ...]
+        current, current_size = [], 0
+        for pid in sorted(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: an
+        # 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:
+                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
+            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 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()
 
@@ -1089,6 +1216,23 @@ 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 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()
         return self.store.move(name, new_name, delete=delete, undelete=undelete, deleted=deleted)