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,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 +321,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 +331,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 +381,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 +412,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 +426,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()

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 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
+
 # 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
 
@@ -10,6 +11,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 +390,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 +554,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 +1061,112 @@ 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 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.
+        """
+        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 and
+        # 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[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, 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 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: 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), 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 = []
+            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
+        # 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()
 
@@ -1089,6 +1205,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)