put repo objects as parts, join them into the pack once, refs #10058

src/borg/cache.py

@@ -1264,10 +1264,11 @@ def add_chunk(
             # there could be a long time without any repository operations and the repo lock would get stale.
             self.refresh_lock(now)
             return self.reuse_chunk(id, size, stats)
-        cdata = self.repo_objs.format(
+        # the parts may reference data (see format_parts), put() copies what it needs to keep.
+        cdata_parts = self.repo_objs.format_parts(
             id, meta, data, compress=compress, size=size, ctype=ctype, clevel=clevel, ro_type=ro_type
         )
-        pack_results = self.repository.put(id, cdata)
+        pack_results = self.repository.put(id, cdata_parts)
         self.last_refresh_dt = now  # .put also refreshed the lock
         self.chunks.add(id, size)
         self.chunks.update_pack_info(pack_results)

src/borg/crypto/key.py

@@ -425,6 +425,14 @@ def id_hash(self, data):
     def encrypt(self, id, data, aad=b""):
         pass
 
+    def encrypt_parts(self, id, data, aad=b""):
+        """Like encrypt(), but return the envelope as a list of parts, see RepoObj.format_parts.
+
+        The parts joined are what encrypt() returns. A part may be *data* itself (not copied), so it
+        is only valid as long as the caller's *data* buffer is.
+        """
+        return [self.encrypt(id, data, aad=aad)]
+
     def encrypt_oneshot(self, id, data, aad=b""):
         """Like encrypt(), but safe to call from any thread, concurrently with encrypt(), see AEADKeyBase."""
         raise NotImplementedError
@@ -1134,9 +1142,14 @@ def _tag_prefix(self, header: bytes, aad: bytes, id: bytes) -> bytes:
         return b"".join([header, len(aad_full).to_bytes(2, "big"), aad_full])
 
     def encrypt(self, id, data, aad=b""):
+        return b"".join(self.encrypt_parts(id, data, aad=aad))
+
+    def encrypt_parts(self, id, data, aad=b""):
+        # the payload is not encrypted, so the envelope is the header + tag, followed by data itself:
+        # returning data as a separate part avoids copying the whole payload just to prepend 34 bytes.
         header = self.TYPE_STR + b"\0"  # TYPE + reserved
         tag = self.mac(self._tag_prefix(header, aad, id), data)
-        return b"".join([header, tag, data])
+        return [header + tag, data]
 
     def encrypt_oneshot(self, id, data, aad=b""):
         # encrypt() has no state (it just computes a MAC), so it is safe to call from any thread already.

src/borg/repoobj.py

@@ -128,6 +128,30 @@ def format(
         clevel: int = None,
         ro_type: str = None,
     ) -> bytes:
+        """Build a repo object from metadata and data (maybe compress, then encrypt), see format_parts."""
+        return b"".join(
+            self.format_parts(id, meta, data, compress=compress, size=size, ctype=ctype, clevel=clevel, ro_type=ro_type)
+        )
+
+    def format_parts(
+        self,
+        id: bytes,
+        meta: dict,
+        data: bytes,
+        compress: bool = True,
+        size: int = None,
+        ctype: int = None,
+        clevel: int = None,
+        ro_type: str = None,
+    ) -> list[bytes | memoryview]:
+        """Like format(), but return the repo object as a list of parts that, joined, give the object.
+
+        This avoids copying the encrypted data just to concatenate it with the header and the metadata:
+        Repository.put accepts the parts and the PackWriter joins them into the pack once (#10058).
+
+        A part may be *data* itself (e.g. with an "authenticated-*" key and data that did not get
+        compressed), so the parts are only valid as long as the caller's *data* buffer is.
+        """
         assert isinstance(ro_type, str)
         assert ro_type != ROBJ_DONTCARE
         meta["type"] = ro_type
@@ -149,12 +173,13 @@ def format(
             data_compressed = data  # is already compressed, is NOT prefixed by type/level bytes
             meta["csize"] = len(data_compressed)
         header_aad = OBJ_MAGIC + bytes([OBJ_VERSION]) + id
-        data_encrypted = self.key.encrypt(id, data_compressed, aad=header_aad + DATA_AAD_TAG)
+        data_parts = self.key.encrypt_parts(id, data_compressed, aad=header_aad + DATA_AAD_TAG)
         meta_packed = msgpack.packb(meta)
         meta_encrypted = self.key.encrypt(id, meta_packed, aad=header_aad + META_AAD_TAG)
-        hdr = self.ObjHeader(OBJ_MAGIC, OBJ_VERSION, id, len(meta_encrypted), len(data_encrypted))
+        data_size = sum(len(part) for part in data_parts)
+        hdr = self.ObjHeader(OBJ_MAGIC, OBJ_VERSION, id, len(meta_encrypted), data_size)
         hdr_packed = self.obj_header.pack(*hdr)
-        return hdr_packed + meta_encrypted + data_encrypted
+        return [hdr_packed + meta_encrypted, *data_parts]
 
     def parse_meta(self, id: bytes, cdata: bytes | memoryview, ro_type: str) -> dict:
         # when calling parse_meta, enough cdata needs to be supplied to contain completely the

src/borg/repository.py

@@ -153,7 +153,7 @@ def build_rest_backend(location):
 class PackWriter:
     """Buffers chunks into a pack file and writes it to the store when full.
 
-    add() buffers a (chunk_id, cdata) pair and marks the chunk pending (F_PENDING), unless
+    add() buffers a (chunk_id, parts) pair and marks the chunk pending (F_PENDING), unless
     its index entry already has a resolved pack location (a re-put of a stored chunk keeps
     that location, so the chunk stays readable, also if an abort drops the buffered copy);
     when the pack is full, it is built, hashed and stored, and each entry's pack_id,
@@ -200,7 +200,7 @@ def __init__(self, store, *, max_count=None, max_size=None, chunks=None, reposit
         # BORG_PACK_TRACE=yes prints store-thread lifecycle markers to stderr, see _trace.
         self.trace_store = os.environ.get("BORG_PACK_TRACE", "no") == "yes"
         self._chunks = chunks  # used when there is no repository
-        self._pieces = []  # list of (chunk_id, cdata)
+        self._pieces = []  # list of (chunk_id, parts): parts is a tuple of bytes, joined they are the object
         self._size = 0  # byte size of buffered pieces
         self._inflight = None  # (thread, outcome) of the pack store-thread, at most one in flight
 
@@ -215,14 +215,21 @@ def chunks(self):
     def add(self, chunk_id, cdata):
         """Buffer a chunk.
 
+        cdata is the repo object as bytes or as a list of parts (see RepoObj.format_parts). The pack
+        is only built later, so a part that is not bytes (e.g. a memoryview into a buffer the caller
+        reuses) is copied here; bytes parts are buffered as they are.
+
         When the chunk fills the pack, the pack is written and the results of the
         *previously* written pack (with async_store) or of this pack (without) are
         returned as a list of (chunk_id, pack_id, obj_offset, obj_size) tuples.
         Returns None when there is nothing to report.
         """
+        if not isinstance(cdata, list):
+            cdata = [cdata]
+        parts = tuple(part if isinstance(part, bytes) else bytes(part) for part in cdata)
         self.chunks.add(chunk_id, 0)  # size: plaintext chunk size, set by the cache layer
-        self._pieces.append((chunk_id, cdata))
-        self._size += len(cdata)
+        self._pieces.append((chunk_id, parts))
+        self._size += sum(len(part) for part in parts)
         if (self.max_count is not None and len(self._pieces) >= self.max_count) or (
             self.max_size is not None and self._size >= self.max_size
         ):
@@ -257,9 +264,10 @@ def _store_pieces(self, pieces, outcome, trace=False):
         """
         self._trace("<", trace)
         try:
-            # Build the pack bytes once by joining all pieces (avoids O(n^2) copies
-            # that incremental string concatenation would cause in Python).
-            pack_data = b"".join(cdata for _, cdata in pieces)
+            # Build the pack bytes once by joining all parts of all pieces (avoids O(n^2) copies
+            # that incremental string concatenation would cause in Python, and the per-object
+            # copies that joining each object's parts first would cause).
+            pack_data = b"".join([part for _, parts in pieces for part in parts])
 
             # Name the pack by the store hash of its bytes: the name commits to the stored content,
             # so borgstore can verify and cache the file.
@@ -269,8 +277,8 @@ def _store_pieces(self, pieces, outcome, trace=False):
             # Record (chunk_id, pack_id, obj_offset, obj_size) for every piece.
             results = []
             offset = 0
-            for chunk_id, cdata in pieces:
-                obj_size = len(cdata)
+            for chunk_id, parts in pieces:
+                obj_size = sum(len(part) for part in parts)
                 results.append((chunk_id, pack_id, offset, obj_size))
                 offset += obj_size
 
@@ -2205,14 +2213,17 @@ def _gather_batch(self, ids, raise_missing):
     def put(self, id, data):
         """put a repo object
 
+        data is the repo object as bytes or as a list of parts (see RepoObj.format_parts), which
+        put() does not keep a reference to beyond its return (see PackWriter.add).
+
         Buffers the chunk in the pack writer.  When the chunk fills the pack and
         triggers a pack write, returns a list of (chunk_id, pack_id, obj_offset, obj_size)
         tuples, one per written chunk; otherwise returns None.  With the background
         store-thread (see PackWriter), the returned tuples are those of the *previous*
         pack, whose store was joined before handing off the current one.
         """
         self._lock_refresh()
-        data_size = len(data)
+        data_size = sum(len(part) for part in data) if isinstance(data, list) else len(data)
         if data_size > MAX_DATA_SIZE:
             raise IntegrityError(f"More than allowed put data [{data_size} > {MAX_DATA_SIZE}]")
         # PackWriter shares this repository's index, so add() triggers the lazy build itself.

src/borg/testsuite/repoobj_test.py

@@ -69,6 +69,30 @@ def test_format_parse_roundtrip(key):
     assert data == got_data
 
 
+@pytest.mark.parametrize("key_class", [AuthenticatedKey, CHPOKey, AESOCBKey])
+@pytest.mark.parametrize("compression", ["none", "lz4"])
+def test_format_parts(key_class, compression):
+    # the parts joined are a valid object, the same format() returns.
+    key = key_class(None)
+    key.init_from_random_data()
+    key.init_ciphers()
+    repo_objs = RepoObj(key)
+    repo_objs.compressor = CompressionSpec(compression).compressor
+    data = b"foobar" * 100
+    id = repo_objs.id_hash(data)
+    parts = repo_objs.format_parts(id, {}, memoryview(data), ro_type=ROBJ_FILE_STREAM)
+    cdata = b"".join(parts)
+    got_meta, got_data = repo_objs.parse(id, cdata, ro_type=ROBJ_FILE_STREAM)
+    assert got_data == data
+    assert got_meta["size"] == len(data)
+    if key_class is AuthenticatedKey:
+        # deterministic envelope (no IV): format() gives the same bytes.
+        assert repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM) == cdata
+        if compression == "none":
+            # the not encrypted, not compressed payload is a part of its own, data is not copied.
+            assert parts[-1].obj is data
+
+
 def test_format_parse_roundtrip_borg1(legacy_key):  # legacy
     repo_objs = RepoObj1(legacy_key)
     data = b"foobar" * 10

src/borg/testsuite/repository_test.py

@@ -1419,6 +1419,33 @@ def test_max_data_size(repo_fixtures, request):
         repository.delete(H(0), validate=None)
 
 
+def test_max_data_size_parts(repo_fixtures, request):
+    # for an object given as a list of parts, the limit applies to the parts' total size.
+    with get_repository_from_fixture(repo_fixtures, request) as repository:
+        max_data = b"x" * (MAX_DATA_SIZE - RepoObj.obj_header.size)
+        with pytest.raises(IntegrityError):
+            repository.put(H(1), [fchunk(max_data), b"x"])
+
+
+def test_put_parts(tmp_path):
+    # put() takes an object as a list of parts (see RepoObj.format_parts): it stores the joined parts.
+    # parts that are not bytes are copied by put(), so the caller may reuse their buffer right after.
+    chunk0, chunk1 = fchunk(b"DATA0", chunk_id=H(0)), fchunk(b"DATA1", chunk_id=H(1))
+    buffer = bytearray(chunk1)
+    with Repository(os.fspath(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        repository.put(H(0), [chunk0[:10], memoryview(chunk0)[10:20], chunk0[20:]])
+        with memoryview(buffer) as view:
+            repository.put(H(1), [view[:30], view[30:]])
+        buffer[:] = b"\xff" * len(buffer)  # the caller reuses its buffer before the pack is written
+        repository.flush()
+        assert repository.get(H(0)) == chunk0
+        assert repository.get(H(1)) == chunk1
+        entry0, entry1 = repository.chunks[H(0)], repository.chunks[H(1)]
+        assert entry0.pack_id == entry1.pack_id
+        assert (entry0.obj_offset, entry0.obj_size) == (0, len(chunk0))
+        assert (entry1.obj_offset, entry1.obj_size) == (len(chunk0), len(chunk1))
+
+
 def check(repository, repo_path, repair=False, status=True):
     assert repository.check(repair=repair) == status
     # Make sure no tmp files are left behind