re-add repo-compress, fixes #9663

reimplemented for the pack-based repository format: the repository is
processed pack after pack - each pack is read as a whole and, if it has
objects not stored with the desired compression, those get recompressed
and the pack is rewritten (objects already fine are copied into the new
pack unchanged), the chunk index is updated and the old pack is deleted.
a pack whose objects all already match is not touched at all.

- new Repository.transform_pack: rewrite one pack, passing each indexed
  object's bytes through a transform callback. gap bytes are handled
  like in compact_pack: superseded duplicates are dropped, other
  unindexed bytes are carried forward for "borg check --repair".
- factored the check_pack_objects / superseded_gap_ranges helpers out
  of compact_pack, now shared with merge_packs and transform_pack.
- crash safety like compact (#9748): stored chunk indexes are
  invalidated before the first store change, a full updated index is
  written back at the end. SIGINT stops cleanly at a pack boundary,
  saving a valid index; a later run recompresses the remaining packs.
- --stats reports rewritten packs, per-object outcomes (recompressed /
  already had the desired compression / kept as-is because
  recompression brings no gain) and the repository size change.
- re-adding the fish completions also restores the compression_methods
  definition that the create/transfer/recreate/import-tar completions
  kept referencing after it was removed together with the old
  repo-compress.

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
Thomas Waldmann
2026-08-02 10:12:06 +02:00
co-authored by Claude Fable 5
parent 2923ac6421
commit 637972f363
10 changed files with 806 additions and 67 deletions
+2
View File
@@ -464,6 +464,8 @@ are calculated *before* compression. New compression settings
will only be applied to new chunks, not existing chunks. So it's safe
to change them.
Use ``borg repo-compress`` to efficiently recompress a complete repository.
Why is backing up an unmodified FAT filesystem slow on Linux?
-------------------------------------------------------------
+1
View File
@@ -40,6 +40,7 @@ Usage
usage/repo-space
usage/repo-list
usage/repo-info
usage/repo-compress
usage/repo-delete
usage/serve
usage/version
+12
View File
@@ -0,0 +1,12 @@
.. include:: repo-compress.rst.inc
Examples
~~~~~~~~
::
# Recompress repository contents
$ borg repo-compress --progress --compression=zstd,3
# Recompress and obfuscate repository contents
$ borg repo-compress --progress --compression=obfuscate,1,zstd,3
+76
View File
@@ -0,0 +1,76 @@
.. IMPORTANT: this file is auto-generated from borg's built-in help, do not edit!
.. _borg_repo-compress:
borg repo-compress
------------------
.. code-block:: none
borg [common options] repo-compress [options]
.. only:: html
.. class:: borg-options-table
+-------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------------------+
| **options** |
+-------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------------------+
| | ``-C COMPRESSION``, ``--compression COMPRESSION`` | select compression algorithm, see the output of the "borg help compression" command for details. |
+-------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------------------+
| | ``-s``, ``--stats`` | print statistics |
+-------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------------------+
| .. class:: borg-common-opt-ref |
| |
| :ref:`common_options` |
+-------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------------------+
.. raw:: html
<script type='text/javascript'>
$(document).ready(function () {
$('.borg-options-table colgroup').remove();
})
</script>
.. only:: latex
options
-C COMPRESSION, --compression COMPRESSION select compression algorithm, see the output of the "borg help compression" command for details.
-s, --stats print statistics
:ref:`common_options`
|
Description
~~~~~~~~~~~
Repository (re-)compression (and/or re-obfuscation).
Reads all repository objects and recompresses the ones that are not already using
the compression type/level and obfuscation level given via ``--compression``.
The repository is processed one pack file at a time: a pack is read as a whole and,
if it holds objects that need recompression, rewritten as a whole - objects already
using the desired compression are copied into the rewritten pack unchanged. A pack
whose objects all already use the desired compression is not touched at all.
Please note that the outcome of recompressing a chunk might not always be the
desired compression type/level - if no compression gives a shorter output, that
might be chosen; such chunks are kept as they are.
Rewriting a pack invalidates every client's cached chunk index, so the next borg
operation of each client will re-fetch the chunk index from the repository.
This command needs free space in the repository for the rewritten pack files
(roughly one pack file size while running).
If the ``borg repo-compress`` process receives a SIGINT signal (Ctrl-C), it stops
at the next pack boundary, leaving the repository and its chunk index in a
consistent state; running it again later processes the remaining packs.
Both ``--progress`` and ``--stats`` are recommended when ``borg repo-compress``
is used interactively.
You do **not** need to run ``borg compact`` after ``borg repo-compress``.
+7
View File
@@ -20,6 +20,7 @@ complete -c borg -f -n __fish_is_first_token -a 'compact' -d 'Free repository sp
complete -c borg -f -n __fish_is_first_token -a 'info' -d 'Show archive details'
complete -c borg -f -n __fish_is_first_token -a 'mount' -d 'Mount archive or a repository'
complete -c borg -f -n __fish_is_first_token -a 'umount' -d 'Unmount the mounted archive'
complete -c borg -f -n __fish_is_first_token -a 'repo-compress' -d 'Repository (re-)compression'
complete -c borg -f -n __fish_is_first_token -a 'repo-create' -d 'Create a new, empty repository'
complete -c borg -f -n __fish_is_first_token -a 'repo-delete' -d 'Delete a repository'
complete -c borg -f -n __fish_is_first_token -a 'repo-info' -d 'Show repository information'
@@ -98,6 +99,12 @@ complete -c borg -f -l 'newest' -d 'Consider archives within
complete -c borg -f -l 'older' -d 'Consider archives older than TIMESPAN' -n "__fish_seen_subcommand_from analyze"
complete -c borg -f -l 'newer' -d 'Consider archives newer than TIMESPAN' -n "__fish_seen_subcommand_from analyze"
# borg repo-compress options
# Define compression methods once at the top
set -l compression_methods "none auto lz4 zstd,1 zstd,2 zstd,3 zstd,4 zstd,5 zstd,6 zstd,7 zstd,8 zstd,9 zstd,10 zstd,11 zstd,12 zstd,13 zstd,14 zstd,15 zstd,16 zstd,17 zstd,18 zstd,19 zstd,20 zstd,21 zstd,22 zlib,1 zlib,2 zlib,3 zlib,4 zlib,5 zlib,6 zlib,7 zlib,8 zlib,9 lzma,0 lzma,1 lzma,2 lzma,3 lzma,4 lzma,5 lzma,6 lzma,7 lzma,8 lzma,9"
complete -c borg -f -s C -l 'compression' -d 'Select compression ALGORITHM,LEVEL [lz4]' -a "$compression_methods" -n "__fish_seen_subcommand_from repo-compress"
complete -c borg -f -s s -l 'stats' -d 'Print statistics' -n "__fish_seen_subcommand_from repo-compress"
# borg create options
complete -c borg -f -s n -l 'dry-run' -d 'Do not create a backup archive' -n "__fish_seen_subcommand_from create"
complete -c borg -f -s s -l 'stats' -d 'Print verbose statistics' -n "__fish_seen_subcommand_from create"
+3
View File
@@ -81,6 +81,7 @@ from .list_cmd import ListMixIn
from .lock_cmds import LocksMixIn
from .mount_cmds import MountMixIn
from .prune_cmd import PruneMixIn
from .repo_compress_cmd import RepoCompressMixIn
from .recreate_cmd import RecreateMixIn
from .rename_cmd import RenameMixIn
from .repo_create_cmd import RepoCreateMixIn
@@ -117,6 +118,7 @@ class Archiver(
PruneMixIn,
RecreateMixIn,
RenameMixIn,
RepoCompressMixIn,
RepoCreateMixIn,
RepoDeleteMixIn,
RepoInfoMixIn,
@@ -305,6 +307,7 @@ class Archiver(
self.build_parser_locks(subparsers, common_parser, mid_common_parser)
self.build_parser_mount_umount(subparsers, common_parser, mid_common_parser)
self.build_parser_prune(subparsers, common_parser, mid_common_parser)
self.build_parser_repo_compress(subparsers, common_parser, mid_common_parser)
self.build_parser_repo_create(subparsers, common_parser, mid_common_parser)
self.build_parser_repo_delete(subparsers, common_parser, mid_common_parser)
self.build_parser_repo_info(subparsers, common_parser, mid_common_parser)
+234
View File
@@ -0,0 +1,234 @@
import logging
from collections import defaultdict
from borgstore.store import ItemInfo
from ._common import with_repository, Highlander
from ..cache import build_chunkindex_from_repo, delete_chunkindex_from_repo, write_chunkindex_to_repo
from ..compress import ObfuscateSize, Auto, COMPRESSOR_TABLE
from ..constants import * # NOQA
from ..helpers import sig_int, ProgressIndicatorPercent, Error, CompressionSpec
from ..helpers import format_file_size, hex_to_bin
from ..helpers.argparsing import ArgumentParser
from ..manifest import Manifest
from ..repository import Repository
from ..logger import create_logger
logger = create_logger()
def get_csettings(c):
"""Return the (ctype, clevel, olevel) compression settings a compressor is configured for."""
if isinstance(c, Auto):
return get_csettings(c.compressor)
if isinstance(c, ObfuscateSize):
ctype, clevel, _ = get_csettings(c.compressor)
olevel = c.level
return ctype, clevel, olevel
ctype, clevel, olevel = c.ID, c.level, -1
return ctype, clevel, olevel
def format_compression_spec(ctype, clevel, olevel):
obfuscation = "" if olevel == -1 else f"obfuscate,{olevel},"
for cname, cls in COMPRESSOR_TABLE.items():
if cls.ID == ctype:
cname = f"{cname}"
break
else:
cname = f"{ctype}"
clevel = f",{clevel}" if clevel != 255 else ""
return obfuscation + cname + clevel
class PackRecompressor:
"""Recompress a repository's objects to the desired compression, one pack at a time.
Each pack is loaded once (via Repository.transform_pack); objects not stored with the desired
compression get recompressed, the others are carried into the rewritten pack unchanged. A pack
whose objects all already match is not rewritten at all.
"""
def __init__(self, repository, manifest, *, print_stats):
self.repository = repository
assert isinstance(repository, Repository)
self.repo_objs = manifest.repo_objs
self.print_stats = print_stats
self.wanted = get_csettings(self.repo_objs.compressor) # desired compression set by --compression
self.chunks = None # a ChunkIndex: chunk id -> pack location
self.store_changed = False # True once the store (and thus any stored chunk index) was modified
# per-object outcomes; the three counters are disjoint, their sum is the objects seen
self.objects_ok = 0 # already stored with the desired compression
self.objects_kept = 0 # recompression yielded what is already stored -> old object kept
self.objects_recompressed = 0 # recompressed and rewritten
self.packs_count = 0
self.packs_rewritten = 0
def recompress(self):
"""Recompress the repository pack after pack; Ctrl-C stops cleanly at a pack boundary."""
logger.info(f"Recompressing repository to {format_compression_spec(*self.wanted)}...")
self.chunks = build_chunkindex_from_repo(self.repository, write_immediately=True)
# group the indexed objects per pack; transform_pack requires each pack's complete id list.
per_pack = defaultdict(list) # pack_id -> [chunk_id, ...]
for id, entry in self.chunks.iteritems():
per_pack[entry.pack_id].append(id)
# the pack files actually in the store; sorted, so the processing order is reproducible.
packs = sorted(
(hex_to_bin(info.name), info.size) for info in map(ItemInfo._make, self.repository.store_list("packs"))
)
self.packs_count = len(packs)
size_before = sum(size for _, size in packs)
size_after = size_before
stale_packs = set(per_pack) - {pack_id for pack_id, _ in packs}
if stale_packs:
# keep these entries: they may be an archive's only pointer to a chunk. dropping them is
# "borg check --repair"'s call.
stale = sum(len(per_pack[pack_id]) for pack_id in stale_packs)
logger.warning(f'{stale} index entries reference missing pack files; run "borg check --repair".')
pi = ProgressIndicatorPercent(
total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress"
)
for i, (pack_id, pack_size) in enumerate(packs):
if sig_int:
break # stop cleanly at a pack boundary: save the index below, then raise
ids = per_pack.get(pack_id)
# a pack without indexed objects (all-gap) is left for "borg check --repair", see #9868.
if ids:
new_pack_id, new_size = self.repository.transform_pack(
pack_id, ids, self.transform, chunks=self.chunks, before_change=self.invalidate_stored_index
)
if new_pack_id != pack_id:
self.packs_rewritten += 1
size_after += new_size - pack_size
pi.show(i + 1) # report after the work, so the final pack lands on 100%
pi.finish()
if self.store_changed:
# the stored chunk indexes were invalidated (invalidate_stored_index), so write a full
# updated index back to the repo; entries were repointed by transform_pack along the way.
write_chunkindex_to_repo(
self.repository, self.chunks, incremental=False, clear=True, force_write=True, delete_other=True
)
self.chunks = None # nothing there (cleared!)
if sig_int:
# Ctrl-C / SIGINT: raise after saving, so an interrupted run still leaves a valid index
raise Error("Got Ctrl-C / SIGINT.")
if self.print_stats:
self.report(size_before, size_after)
def invalidate_stored_index(self):
"""Called by transform_pack just before its first store change (crash-safety, see #9748)."""
if not self.store_changed:
delete_chunkindex_from_repo(self.repository)
self.store_changed = True
def transform(self, id, obj_bytes):
"""Return the recompressed object, or obj_bytes unchanged if recompression is not needed."""
meta = self.repo_objs.parse_meta(id, obj_bytes, ro_type=ROBJ_DONTCARE)
found = meta["ctype"], meta["clevel"], meta.get("olevel", -1)
if found == self.wanted:
self.objects_ok += 1
return obj_bytes
meta, data = self.repo_objs.parse(id, obj_bytes, ro_type=ROBJ_DONTCARE)
ro_type = meta.pop("type", None)
if self.wanted[2] == -1:
# if the object was obfuscated, but should not be in future, remove related metadata
meta.pop("olevel", None)
meta.pop("psize", None)
new_bytes = self.repo_objs.format(id, meta, data, ro_type=ro_type)
# format() filled meta with the compression actually done - not always the desired one: a
# DecidingCompressor stores data that does not compress well differently than told.
done = meta["ctype"], meta["clevel"], meta.get("olevel", -1)
if done == found:
# the outcome is what is already stored: keep the old object, otherwise such objects
# would get recompressed and rewritten **again and again** with no gain.
self.objects_kept += 1
return obj_bytes
self.objects_recompressed += 1
return new_bytes
def report(self, size_before, size_after):
# "borg.output.stats" is enabled at INFO level whenever --stats is given (implied logging).
stats_logger = logging.getLogger("borg.output.stats")
objects = self.objects_ok + self.objects_kept + self.objects_recompressed
stats_logger.info("Recompression stats:")
stats_logger.info(f"Packs: {self.packs_count} total, {self.packs_rewritten} rewritten.")
stats_logger.info(
f"Objects: {objects} total, {self.objects_recompressed} recompressed, "
f"{self.objects_ok} already had the desired compression, "
f"{self.objects_kept} kept as-is (recompression brings no gain)."
)
delta = size_before - size_after
change = "shrunk" if delta >= 0 else "grew"
stats_logger.info(
f"Repository size: {format_file_size(size_before)} before, {format_file_size(size_after)} after, "
f"{change} by {format_file_size(abs(delta))}."
)
class RepoCompressMixIn:
@with_repository(manifest=True, exclusive=True, compatibility=(Manifest.Operation.CHECK,))
def do_repo_compress(self, args, repository, manifest):
"""Repository (re-)compression."""
if not isinstance(repository, Repository):
raise Error("repo-compress not supported for legacy repositories.")
# refuse up front on a repo opened read-only, before a fully-compliant repo could make
# recompression a silent no-op.
repository.assert_writable()
PackRecompressor(repository, manifest, print_stats=args.stats).recompress()
def build_parser_repo_compress(self, subparsers, common_parser, mid_common_parser):
from ._common import process_epilog
repo_compress_epilog = process_epilog(
"""
Repository (re-)compression (and/or re-obfuscation).
Reads all repository objects and recompresses the ones that are not already using
the compression type/level and obfuscation level given via ``--compression``.
The repository is processed one pack file at a time: a pack is read as a whole and,
if it holds objects that need recompression, rewritten as a whole - objects already
using the desired compression are copied into the rewritten pack unchanged. A pack
whose objects all already use the desired compression is not touched at all.
Please note that the outcome of recompressing a chunk might not always be the
desired compression type/level - if no compression gives a shorter output, that
might be chosen; such chunks are kept as they are.
Rewriting a pack invalidates every client's cached chunk index, so the next borg
operation of each client will re-fetch the chunk index from the repository.
This command needs free space in the repository for the rewritten pack files
(roughly one pack file size while running).
If the ``borg repo-compress`` process receives a SIGINT signal (Ctrl-C), it stops
at the next pack boundary, leaving the repository and its chunk index in a
consistent state; running it again later processes the remaining packs.
Both ``--progress`` and ``--stats`` are recommended when ``borg repo-compress``
is used interactively.
You do **not** need to run ``borg compact`` after ``borg repo-compress``.
"""
)
subparser = ArgumentParser(
parents=[common_parser], description=self.do_repo_compress.__doc__, epilog=repo_compress_epilog
)
subparsers.add_subcommand("repo-compress", subparser, help=self.do_repo_compress.__doc__)
subparser.add_argument(
"-C",
"--compression",
metavar="COMPRESSION",
dest="compression",
type=CompressionSpec,
default=CompressionSpec("lz4"),
action=Highlander,
help="select compression algorithm, see the output of the " '"borg help compression" command for details.',
)
subparser.add_argument("-s", "--stats", dest="stats", action="store_true", help="print statistics")
+175 -67
View File
@@ -248,6 +248,70 @@ class PackReader:
offset += obj_size
def check_pack_objects(pack_hex, obj_ranges, pack_size):
"""Validate a pack's indexed objects against the pack's file size.
obj_ranges: the offset-ordered (obj_offset, obj_size) ranges of the pack's indexed objects.
An overlap between objects, or an object claiming to end past the pack file (pack_size bytes),
means index corruption and raises IntegrityError.
"""
covered = 0
for offset, size in obj_ranges:
if offset < covered:
raise IntegrityError(
f'pack {pack_hex}: overlapping objects at offset {offset} (index corruption), run "borg check"'
)
covered = offset + size
if covered > pack_size:
raise IntegrityError(
f'pack {pack_hex}: object extends past end of file at offset {covered} (index corruption), run "borg check"'
)
def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
"""Find the superseded duplicates among a pack's gap bytes (bytes no index entry covers).
A gap holds a chunk copy stored again elsewhere, or objects from a backup that crashed before
writing its index. Walk each gap's object headers: an object whose chunk id the index maps to a
different location is a superseded duplicate (the id is a keyed MAC of the plaintext, so equal
ids mean equal content) and its bytes are redundant. An object whose id is not in the index
(borg check --repair re-indexes it) or whose entry points back at this offset (its only copy)
is not reported. A header that does not parse or overruns its gap ends the walk over that gap.
obj_ranges: the offset-ordered, validated (obj_offset, obj_size) ranges of the pack's indexed
objects; the gaps are the byte ranges between (and after) them.
Returns the offset-ordered list of (offset, size) ranges holding superseded duplicates.
"""
# find the gaps: byte ranges no indexed object covers.
gaps = [] # (start, end) of each gap, offset-ordered
cursor = 0
for offset, size in obj_ranges:
if offset > cursor:
gaps.append((cursor, offset))
cursor = offset + size
if cursor < pack_size:
gaps.append((cursor, pack_size))
drop_ranges = [] # (obj_offset, obj_size) of superseded duplicates, offset-ordered
hdr_size = RepoObj.obj_header.size
for gstart, gend in gaps:
offset = gstart
while offset < gend:
hdr_data = reader.read(offset, hdr_size)
if len(hdr_data) < hdr_size:
break
hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
obj_size = hdr_size + hdr.meta_size + hdr.data_size
if hdr.magic != OBJ_MAGIC or offset + obj_size > gend:
break
if hdr.chunk_id in chunks:
entry = chunks[hdr.chunk_id]
if entry.pack_id != pack_id or entry.obj_offset != offset:
drop_ranges.append((offset, obj_size))
offset += obj_size
return drop_ranges
class PackTracker:
"""Packs verified in the current check cycle, mapping pack_id -> (timestamp, result).
@@ -1045,65 +1109,21 @@ class Repository:
located.append((entry.obj_offset, obj_id, entry.obj_size, keep))
located.sort()
# walk objects in offset order. covered is the end offset of the last object; an overlap
# (offset < covered) is index corruption. record the dropped objects' byte ranges; every other
# byte (kept objects and gaps that no index entry covers) is copied into the new pack unchanged.
drop_ranges = [] # (obj_offset, obj_size) of dropped objects, offset-ordered
covered = 0
for offset, obj_id, size, keep in located:
if offset < covered:
raise IntegrityError(
f"pack {bin_to_hex(pack_id)}: overlapping objects at offset {offset} (index corruption), "
f'run "borg check"'
)
covered = offset + size
if not keep:
drop_ranges.append((offset, size))
# reject an object that ends past the pack file: store.defrag would short-read it into a
# truncated object in the new pack, then the intact source pack is deleted.
# validate the listed objects. an overlap is index corruption; so is an object ending past
# the pack file: store.defrag would short-read it into a truncated object in the new pack,
# then the intact source pack is deleted.
pack_size = self.store.info(pack_key).size
if covered > pack_size:
raise IntegrityError(
f"pack {bin_to_hex(pack_id)}: object extends past end of file at offset {covered} "
f'(index corruption), run "borg check"'
)
obj_ranges = [(offset, size) for offset, _, size, _ in located]
check_pack_objects(bin_to_hex(pack_id), obj_ranges, pack_size)
# find the gaps: byte ranges no listed object covers (a chunk copy stored again elsewhere, or
# objects from a backup that crashed before writing its index).
gaps = [] # (start, end) of each gap, offset-ordered
cursor = 0
for offset, obj_id, size, keep in located:
if offset > cursor:
gaps.append((cursor, offset))
cursor = offset + size
if cursor < pack_size:
gaps.append((cursor, pack_size))
# walk each gap's object headers. drop an object whose chunk id the index maps to a different
# location: that entry is the authoritative copy (the id is a keyed MAC of the plaintext, so
# equal ids mean equal content) and these bytes are redundant. keep an object whose id is not in
# the index (borg check --repair re-indexes it) or whose entry points back at this offset (its
# only copy). a header that does not parse or overruns its gap ends the walk over that gap.
# record the dropped objects' byte ranges; every other byte (kept objects and gaps that no
# index entry covers) is copied into the new pack unchanged. superseded duplicates found in
# the gaps are dropped along with them (see superseded_gap_ranges).
# TODO(#9868 follow-up): classify gaps in compact_packs pass 1 too, so superseded bytes count
# toward the rewrite threshold and a wholly superseded orphan pack can be dropped outright.
drop_ranges = [(offset, size) for offset, _, size, keep in located if not keep]
reader = PackReader(store=self.store, pack_id=pack_id)
hdr_size = RepoObj.obj_header.size
for gstart, gend in gaps:
offset = gstart
while offset < gend:
hdr_data = reader.read(offset, hdr_size)
if len(hdr_data) < hdr_size:
break
hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
obj_size = hdr_size + hdr.meta_size + hdr.data_size
if hdr.magic != OBJ_MAGIC or offset + obj_size > gend:
break
if hdr.chunk_id in chunks:
entry = chunks[hdr.chunk_id]
if entry.pack_id != pack_id or entry.obj_offset != offset:
drop_ranges.append((offset, obj_size))
offset += obj_size
drop_ranges += superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size)
drop_ranges.sort()
dropped_bytes = sum(size for _, size in drop_ranges) # on-disk bytes this rewrite frees, for --stats
@@ -1207,19 +1227,7 @@ class Repository:
# 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"'
)
check_pack_objects(bin_to_hex(pid), ((offset, size) for offset, _, size in per_pack[pid]), pack_size[pid])
# greedily batch whole pack files so each output pack stays within max_size, in sorted id
# order so batch composition is reproducible.
@@ -1269,6 +1277,106 @@ class Repository:
pi.show(increase=1)
pi.finish()
def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change=None):
"""Rewrite pack <pack_id>, passing each indexed object's bytes through <transform>.
ids: the chunk ids of this pack's objects. Must cover every object the chunk index lists
for this pack (same contract as compact_pack's keep_ids/drop_ids): an unlisted indexed
object would keep its bytes in the new pack, but its index entry would go stale when
the old pack is deleted.
transform: called as transform(chunk_id, obj_bytes) with an object's stored bytes; returns
the replacement bytes, or obj_bytes itself (the identical bytes object) to keep the
object unchanged. The chunk id (and thus the plaintext) must not change; sizes may.
chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index
updates to. Must be the index <ids> was derived from. Default: self.chunks.
before_change: called once, just before the first store modification; use it to invalidate
stored chunk indexes for crash safety (see #9748). Not called when the pack is kept.
The whole pack file is loaded into memory (bounded by the pack size limit). Gap bytes
(bytes no index entry covers) are handled like in compact_pack: an object superseded by a
copy stored elsewhere is dropped, all other unindexed bytes are copied into the new pack
unchanged, to be handled by "borg check --repair". An overlap between indexed objects, or
an object claiming to end past the pack file, means index corruption and raises
IntegrityError, before anything is written.
If every object is kept and no gap bytes are dropped, the store and the chunk index are not
touched at all. Otherwise the new pack (named sha256 of its content) is stored, the indexed
objects are repointed at it, and the old pack is deleted last, so the objects' bytes are
never the only copy.
Returns (new_pack_id, new_size): the rewritten pack's id and byte size, or
(pack_id, <old size>) when the pack was kept unchanged.
Updates the in-memory chunk index only; the caller holds the exclusive lock and writes the
index back to the store afterwards.
"""
self._lock_refresh()
if chunks is None:
chunks = self.chunks
pack_hex = bin_to_hex(pack_id)
pack_key = "packs/" + pack_hex
pack_contents = self.store.load(pack_key)
pack_size = len(pack_contents)
reader = PackReader(pack_id=pack_id, pack_contents=pack_contents)
# collect the listed objects' ranges, ordered by offset, and validate them (see docstring).
located = [] # (obj_offset, obj_id, obj_size)
for obj_id in ids:
entry = chunks[obj_id]
assert entry.pack_id == pack_id, f"{bin_to_hex(obj_id)} is not in pack {pack_hex}"
located.append((entry.obj_offset, obj_id, entry.obj_size))
located.sort()
obj_ranges = [(offset, size) for offset, _, size in located]
check_pack_objects(pack_hex, obj_ranges, pack_size)
drop_ranges = superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size)
# assemble the new pack in offset order: transformed objects, dropped ranges skipped, all
# other bytes copied verbatim. the two range lists never overlap (drops lie in gaps), so a
# single offset-sorted walk over both handles the interleaving.
events = [(offset, size, obj_id) for offset, obj_id, size in located]
events += [(offset, size, None) for offset, size in drop_ranges] # None: a range to drop
events.sort(key=lambda event: event[0])
pieces = [] # byte spans of the new pack, in order
new_locations = [] # (obj_id, new_offset, new_size) of each object
changed = bool(drop_ranges)
cursor = 0 # position in the old pack
new_offset = 0 # position in the new pack
for offset, size, obj_id in events:
if offset > cursor: # verbatim span (gap bytes) before this event
pieces.append(pack_contents[cursor:offset])
new_offset += offset - cursor
if obj_id is not None:
obj_bytes = pack_contents[offset : offset + size]
new_bytes = transform(obj_id, obj_bytes)
if new_bytes is not obj_bytes:
changed = True
pieces.append(new_bytes)
new_locations.append((obj_id, new_offset, len(new_bytes)))
new_offset += len(new_bytes)
# else: a dropped range, skip its bytes
cursor = offset + size
if cursor < pack_size: # verbatim span after the last event
pieces.append(pack_contents[cursor:])
if not changed:
return pack_id, pack_size
pack_data = b"".join(pieces)
new_pack_id = sha256(pack_data).digest()
if new_pack_id == pack_id: # the transforms reproduced the pack byte-identically
return pack_id, pack_size
if before_change is not None:
before_change()
# store the new pack before touching the index or the old pack, so a failure leaves
# everything unchanged.
self.store.store("packs/" + bin_to_hex(new_pack_id), pack_data)
chunks.update_pack_info([(obj_id, new_pack_id, offset, size) for obj_id, offset, size in new_locations])
# delete the old pack last, after the new one is stored and indexed, so the objects' bytes
# are never the only copy.
self.store_delete(pack_key)
return new_pack_id, len(pack_data)
def break_lock(self):
Lock(self.store).break_lock()
@@ -0,0 +1,291 @@
import os
import pytest
from ...constants import * # NOQA
from ...helpers import bin_to_hex, sig_int, Error, CompressionSpec
from ...repository import Repository, PackReader, repo_lister
from ...cache import list_chunkindex_hashes
from ...manifest import Manifest
from ...compress import ZSTD, ZLIB, LZ4, CNONE
from ...archiver.repo_compress_cmd import PackRecompressor
from . import create_regular_file, cmd, RK_ENCRYPTION
from ..repository_test import H, fchunk, pdchunk
def test_repo_compress(archiver):
def check_compression(ctype, clevel, olevel):
"""Check that all chunks in the repo are compressed/obfuscated as expected."""
repository = Repository(archiver.repository_path, exclusive=True)
with repository:
manifest = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
for id, _ in repo_lister(repository, limit=LIST_SCAN_LIMIT):
chunk = repository.get(id, read_data=True)
meta, data = manifest.repo_objs.parse(
id, chunk, ro_type=ROBJ_DONTCARE
) # will also decompress according to metadata
m_olevel = meta.get("olevel", -1)
m_psize = meta.get("psize", -1)
print(bin_to_hex(id), meta["ctype"], meta["clevel"], meta["csize"], meta["size"], m_olevel, m_psize)
# this is not as easy as one thinks due to the DecidingCompressor choosing the smallest of
# (desired compressed, lz4 compressed, not compressed).
assert meta["ctype"] in (ctype, LZ4.ID, CNONE.ID)
assert meta["clevel"] in (clevel, 255) # LZ4 and CNONE have level 255
if olevel != -1: # we expect obfuscation
assert "psize" in meta
assert m_olevel == olevel
else:
assert "psize" not in meta
assert "olevel" not in meta
create_regular_file(archiver.input_path, "file1", size=1024 * 10)
create_regular_file(archiver.input_path, "file2", contents=os.urandom(1024 * 10))
cmd(archiver, "repo-create", RK_ENCRYPTION)
cname, ctype, clevel, olevel = ZLIB.name, ZLIB.ID, 3, -1
cmd(archiver, "create", "test", "input", "-C", f"{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZSTD.name, ZSTD.ID, 1, -1 # change compressor (and level)
cmd(archiver, "repo-compress", "-C", f"{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZSTD.name, ZSTD.ID, 3, -1 # only change level
cmd(archiver, "repo-compress", "-C", f"{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZSTD.name, ZSTD.ID, 3, 110 # only change to obfuscated
cmd(archiver, "repo-compress", "-C", f"obfuscate,{olevel},{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZSTD.name, ZSTD.ID, 3, 112 # only change obfuscation level
cmd(archiver, "repo-compress", "-C", f"obfuscate,{olevel},{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZSTD.name, ZSTD.ID, 3, -1 # change to not obfuscated
cmd(archiver, "repo-compress", "-C", f"{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZLIB.name, ZLIB.ID, 1, -1
cmd(archiver, "repo-compress", "-C", f"auto,{cname},{clevel}")
check_compression(ctype, clevel, olevel)
cname, ctype, clevel, olevel = ZLIB.name, ZLIB.ID, 2, 111
cmd(archiver, "repo-compress", "-C", f"obfuscate,{olevel},auto,{cname},{clevel}")
check_compression(ctype, clevel, olevel)
# data must survive all those recompressions unharmed
cmd(archiver, "check")
def test_repo_compress_stats(archiver):
create_regular_file(archiver.input_path, "file1", size=1024 * 10)
create_regular_file(archiver.input_path, "file2", contents=os.urandom(1024 * 10))
cmd(archiver, "repo-create", RK_ENCRYPTION)
cname, clevel = ZLIB.name, 3
cmd(archiver, "create", "test", "input", "-C", f"{cname},{clevel}")
cname, clevel = ZSTD.name, 1 # change compressor (and level)
output = cmd(archiver, "repo-compress", "-C", f"{cname},{clevel}", "--stats")
assert "Recompression stats:" in output
def test_repo_compress_multiple_packs(archiver, monkeypatch):
# process a repository whose data spans several packs, pack after pack.
monkeypatch.setenv("BORG_PACK_MAX_SIZE", "65536") # tiny packs, so the repo has quite a few
for i in range(8):
# per file: 128 kiB of distinct data that compresses to roughly a quarter of its size,
# so the compressed objects fill several packs and every pack needs recompression.
create_regular_file(archiver.input_path, f"file{i}", contents=os.urandom(16 * 1024) * 8)
cmd(archiver, "repo-create", RK_ENCRYPTION)
cmd(archiver, "create", "test", "input", "-C", "zlib,3")
with Repository(archiver.repository_path, exclusive=True) as repository:
packs_before = {info.name for info in repository.store_list("packs")}
assert len(packs_before) > 2
cmd(archiver, "repo-compress", "-C", "zstd,3", "--stats")
# several packs were rewritten. not necessarily all of them: a pack holding only incompressible
# objects (e.g. an archive chunkids list, stored uncompressed even under -C zlib) is kept as-is.
with Repository(archiver.repository_path, exclusive=True) as repository:
packs_after = {info.name for info in repository.store_list("packs")}
assert len(packs_before - packs_after) >= 2
# no object is left with the old zlib compression
manifest = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
for id, _ in repo_lister(repository, limit=LIST_SCAN_LIMIT):
meta = manifest.repo_objs.parse_meta(id, repository.get(id, read_data=False), ro_type=ROBJ_DONTCARE)
assert meta["ctype"] != ZLIB.ID
# everything still there and consistent?
cmd(archiver, "check")
output = cmd(archiver, "list", "test")
for i in range(8):
assert f"file{i}" in output
def test_repo_compress_second_run_changes_nothing(archiver):
# a second run with the same settings must not rewrite any pack and must leave the stored
# chunk index alone (no invalidation, no rewrite - other clients' cached indexes stay valid).
create_regular_file(archiver.input_path, "file1", size=1024 * 10)
create_regular_file(archiver.input_path, "file2", contents=os.urandom(1024 * 10))
cmd(archiver, "repo-create", RK_ENCRYPTION)
cmd(archiver, "create", "test", "input", "-C", "zlib,3")
cmd(archiver, "repo-compress", "-C", "zstd,3")
with Repository(archiver.repository_path, exclusive=True) as repository:
packs_before = {info.name for info in repository.store_list("packs")}
hashes_before = list_chunkindex_hashes(repository)
output = cmd(archiver, "repo-compress", "-C", "zstd,3", "--stats")
assert ", 0 recompressed" in output # everything already zstd,3 (or "kept as-is"): nothing to rewrite
with Repository(archiver.repository_path, exclusive=True) as repository:
assert {info.name for info in repository.store_list("packs")} == packs_before
assert list_chunkindex_hashes(repository) == hashes_before
def test_repo_compress_soft_interrupt_persists_valid_index(archiver, monkeypatch):
"""One Ctrl-C stops repo-compress at the next pack boundary, saves a chunk index that still
matches the repository, and exits with an error. A later run recompresses the remaining packs."""
monkeypatch.setenv("BORG_PACK_MAX_COUNT", "1") # one object per pack -> several packs to stop between
for i in range(3):
# compressible, distinct data, so every object really gets recompressed (lz4 -> zstd)
create_regular_file(archiver.input_path, f"file{i}", contents=os.urandom(512) * 4)
cmd(archiver, "repo-create", RK_ENCRYPTION)
cmd(archiver, "create", "archive", "input")
with Repository(archiver.repository_path, exclusive=True) as repository:
pack_names_before = {info.name for info in repository.store_list("packs")}
assert len(pack_names_before) >= 2 # need several packs to observe an early stop
manifest = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
manifest.repo_objs.compressor = CompressionSpec("zstd,3").compressor
recompressor = PackRecompressor(repository, manifest, print_stats=False)
original_store_delete = repository.store_delete
calls = []
def store_delete_then_interrupt(name, **kwargs):
original_store_delete(name, **kwargs)
if name.startswith("packs/"): # only pack deletes, not the index invalidation deletes
calls.append(name)
if len(calls) == 1:
sig_int._sig_int_triggered = True # one Ctrl-C after the first pack was rewritten
monkeypatch.setattr(repository, "store_delete", store_delete_then_interrupt)
try:
with pytest.raises(Error, match="Got Ctrl-C"):
recompressor.recompress()
finally:
sig_int._sig_int_triggered = False # reset the global flag for the following tests
# a valid chunk index was persisted and every entry points at a pack that still exists
with Repository(archiver.repository_path, exclusive=True) as repository:
assert list_chunkindex_hashes(repository) != []
pack_names_after = {info.name for info in repository.store_list("packs")}
# one pack was rewritten before the stop, the remaining old packs are still there
assert 0 < len(pack_names_before - pack_names_after) < len(pack_names_before)
for id, entry in repository.chunks.iteritems():
assert bin_to_hex(entry.pack_id) in pack_names_after
# a later run finishes the recompression of the remaining packs
cmd(archiver, "repo-compress", "-C", "zstd,3")
cmd(archiver, "check")
def transform_via(replacements):
"""Make a transform_pack callback replacing the given objects, keeping all others unchanged."""
def transform(chunk_id, obj_bytes):
return replacements.get(chunk_id, obj_bytes)
return transform
def test_transform_pack_keeps_unindexed_gap(tmp_path):
# bytes no index entry covers (e.g. objects of a backup that crashed before writing its index)
# must be carried into the transformed pack unchanged - recovering them is "borg check --repair"'s
# job. also, the objects around them must be repointed correctly although their sizes changed.
location = os.fspath(tmp_path / "repo")
with Repository(location, exclusive=True, create=True) as repository:
repository._pack_writer.max_count = 3 # one flush() -> one pack
for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX"), (H(2), b"YYYY")]:
repository.put(cid, fchunk(data, chunk_id=cid))
repository.flush()
pack_id = repository.chunks[H(0)].pack_id
del repository.chunks[H(1)] # X's bytes become an unindexed gap between W and Y
replacements = {H(0): fchunk(b"W" * 100, chunk_id=H(0)), H(2): fchunk(b"y", chunk_id=H(2))}
calls = []
new_pack_id, new_size = repository.transform_pack(
pack_id, [H(0), H(2)], transform_via(replacements), before_change=lambda: calls.append(1)
)
assert new_pack_id != pack_id
assert calls == [1] # before_change called (once), the store was modified
# the transformed objects are indexed and readable, with their new sizes
assert pdchunk(repository.get(H(0))) == b"W" * 100
assert pdchunk(repository.get(H(2))) == b"y"
# the old pack is gone, the new one exists with the reported size
pack_names = {info.name: info.size for info in repository.store_list("packs")}
assert bin_to_hex(pack_id) not in pack_names
assert pack_names[bin_to_hex(new_pack_id)] == new_size
# X's object is still in the new pack, as unindexed gap bytes at the right position
gap_objects = [
(chunk_id, offset, size)
for chunk_id, offset, size in PackReader(repository.store, new_pack_id).iter_headers()
if chunk_id == H(1)
]
assert len(gap_objects) == 1
def test_transform_pack_drops_superseded_gap(tmp_path):
# a gap object whose chunk id the index maps to another location is a redundant, superseded
# duplicate (equal ids mean equal content) - a transformed pack must not carry it forward.
location = os.fspath(tmp_path / "repo")
with Repository(location, exclusive=True, create=True) as repository:
repository._pack_writer.max_count = 2 # one flush() -> one pack
# pack A: W and X; pack B: a second copy of X, which repoints the index to pack B,
# leaving X's bytes in pack A as a superseded gap.
for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX")]:
repository.put(cid, fchunk(data, chunk_id=cid))
repository.flush()
pack_a = repository.chunks[H(0)].pack_id
repository.put(H(1), fchunk(b"XXXX", chunk_id=H(1)))
repository.flush()
pack_b = repository.chunks[H(1)].pack_id
assert pack_b != pack_a
w_new = fchunk(b"W" * 100, chunk_id=H(0))
new_pack_id, new_size = repository.transform_pack(pack_a, [H(0)], transform_via({H(0): w_new}))
assert new_pack_id != pack_a
assert new_size == len(w_new) # only W remains, X's superseded bytes were dropped
assert pdchunk(repository.get(H(0))) == b"W" * 100
assert pdchunk(repository.get(H(1))) == b"XXXX" # the authoritative copy in pack B
def test_transform_pack_unchanged_pack_untouched(tmp_path):
# if every transform keeps its object, the store must not be touched at all:
# no pack write, no pack delete, no before_change call.
location = os.fspath(tmp_path / "repo")
with Repository(location, exclusive=True, create=True) as repository:
repository._pack_writer.max_count = 2 # one flush() -> one pack
for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX")]:
repository.put(cid, fchunk(data, chunk_id=cid))
repository.flush()
pack_id = repository.chunks[H(0)].pack_id
pack_names_before = {info.name for info in repository.store_list("packs")}
calls = []
new_pack_id, new_size = repository.transform_pack(
pack_id, [H(0), H(1)], transform_via({}), before_change=lambda: calls.append(1)
)
assert new_pack_id == pack_id
assert calls == [] # nothing changed, so before_change was never called
assert {info.name for info in repository.store_list("packs")} == pack_names_before
assert pdchunk(repository.get(H(0))) == b"WWWW"
assert pdchunk(repository.get(H(1))) == b"XXXX"
@@ -83,6 +83,11 @@ def test_repository_permissions_no_delete(archivers, request, monkeypatch):
# A dry run only reads and reports, so it is allowed even without write/delete access.
cmd(archiver, "compact", "--dry-run")
# Try to repo-compress (and change compression from lz4 to zstd), which should fail up front:
# it rewrites (and deletes) pack files, which is disallowed by no-delete.
with pytest.raises(Repository.PermissionDenied):
cmd(archiver, "repo-compress", "-C", "zstd")
# Check without --repair should work.
cmd(archiver, "check")