diff --git a/src/borg/cache.py b/src/borg/cache.py index fafee9c110..9102ba2582 100644 --- a/src/borg/cache.py +++ b/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) diff --git a/src/borg/crypto/key.py b/src/borg/crypto/key.py index d2c0a8dc1f..d34d23fb63 100644 --- a/src/borg/crypto/key.py +++ b/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. diff --git a/src/borg/repoobj.py b/src/borg/repoobj.py index 750000f245..e6cef72938 100644 --- a/src/borg/repoobj.py +++ b/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 diff --git a/src/borg/repository.py b/src/borg/repository.py index 4ceb987c70..ad9c738198 100644 --- a/src/borg/repository.py +++ b/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,6 +2213,9 @@ 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 @@ -2212,7 +2223,7 @@ def put(self, id, data): 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. diff --git a/src/borg/testsuite/repoobj_test.py b/src/borg/testsuite/repoobj_test.py index 7ca95f0658..f5313e7fe1 100644 --- a/src/borg/testsuite/repoobj_test.py +++ b/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 diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 142630a1ae..ce7c96e37d 100644 --- a/src/borg/testsuite/repository_test.py +++ b/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