From 62f2e8b05f31f8c3cdf454fd50469269fc93b56b Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 26 Jun 2026 01:03:09 +0530 Subject: [PATCH] repository: store more than one object per pack (#9808) * repository: store more than one object per pack Raise PackWriter max_count to 2 and flush buffered packs before the archives pointer is written; also flush after debug put-obj and in the affected tests. * archive: flush buffered packs in Archives.create() so no caller writes the pointer over buffered objects --- src/borg/archiver/debug_cmd.py | 1 + src/borg/manifest.py | 2 + src/borg/repository.py | 2 +- src/borg/testsuite/archiver/check_cmd_test.py | 3 ++ src/borg/testsuite/cache_test.py | 2 + src/borg/testsuite/repository_test.py | 42 ++++++++++++++++++- 6 files changed, 49 insertions(+), 3 deletions(-) diff --git a/src/borg/archiver/debug_cmd.py b/src/borg/archiver/debug_cmd.py index 0ef84804a..3ff831b88 100644 --- a/src/borg/archiver/debug_cmd.py +++ b/src/borg/archiver/debug_cmd.py @@ -281,6 +281,7 @@ class DebugMixIn: raise CommandError(f"object id {hex_id} is invalid [{str(err)}].") repository.put(id, data) + repository.flush() # no cache wraps this command, so flush the buffered pack before close() print("object %s put." % hex_id) @with_repository(manifest=False, exclusive=True) diff --git a/src/borg/manifest.py b/src/borg/manifest.py index efb019485..5c6df5b80 100644 --- a/src/borg/manifest.py +++ b/src/borg/manifest.py @@ -329,6 +329,8 @@ class Archives: if isinstance(ts, datetime): ts = ts.isoformat(timespec="microseconds") assert isinstance(ts, str) + # flush buffered packs first: the pointer must not reference objects still buffered at N>1. + self.repository.flush() # we only create a directory entry, its name points to the archive item: self.repository.store_store(f"archives/{bin_to_hex(id)}", b"") diff --git a/src/borg/repository.py b/src/borg/repository.py index 512f99672..babfd8495 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -507,7 +507,7 @@ class Repository: if lock: self.lock = Lock(self.store, exclusive, timeout=lock_wait).acquire() self._chunks = None - self._pack_writer = PackWriter(self.store, max_count=1, repository=self) + self._pack_writer = PackWriter(self.store, max_count=2, repository=self) self.opened = True @property diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index e3fa7d3df..029d55aea 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -308,6 +308,7 @@ def test_manifest_rebuild_corrupted_chunk(archivers, request): # framing rather than the authenticated payload, which made the repair flaky on Windows. corrupted_chunk = corrupt(chunk, len(chunk) // 2) repository.put(archive.id, corrupted_chunk) + repository.flush() # N>1: make the put durable before close()/the check below cmd(archiver, "check", exit_code=1) output = cmd(archiver, "check", "-v", "--repair", exit_code=0) assert "archive2" in output @@ -366,6 +367,7 @@ def test_spoofed_archive(archivers, request): ro_type=ROBJ_FILE_STREAM, # a real archive is stored with ROBJ_ARCHIVE_META ), ) + repository.flush() # N>1: make the put durable before close()/the check below cmd(archiver, "check", exit_code=1) cmd(archiver, "check", "--repair", "--debug", exit_code=0) output = cmd(archiver, "repo-list") @@ -384,6 +386,7 @@ def test_extra_chunks(archivers, request): key = b"01234567890123456789012345678901" chunk = fchunk(b"xxxx", chunk_id=key) repository.put(key, chunk) + repository.flush() # N>1: make the put durable before close()/the check below cmd(archiver, "check", "-v", exit_code=0) # check does not deal with orphans anymore diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index dee76f8f0..152c022b9 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -20,7 +20,9 @@ class TestAdHocWithFilesCache: self.repository_location = os.path.join(str(tmpdir), "repository") with Repository(self.repository_location, exclusive=True, create=True) as repository: repository.put(H(1), b"1234") + repository.flush() # N>1: the lone put would stay buffered; make it durable yield repository + repository.flush() # flush anything a test buffered via the cache before close() @pytest.fixture def key(self, repository, monkeypatch): diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 74edb40fb..12b0de1fb 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -5,6 +5,7 @@ from hashlib import sha256 import pytest from ..helpers import IntegrityError, Location, bin_to_hex from ..hashindex import ChunkIndex +from ..constants import UNKNOWN_BYTES32 from ..repository import Repository, MAX_DATA_SIZE, rest_serve_command, PackWriter, PackReader from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION from .hashindex_test import H @@ -124,6 +125,7 @@ def test_read_data(repo_fixtures, request): chunk_complete = hdr + meta + data chunk_short = hdr + meta repository.put(H(0), chunk_complete) + repository.flush() # N>1: make the buffered pack durable before get() assert repository.get(H(0)) == chunk_complete assert repository.get(H(0), read_data=True) == chunk_complete assert repository.get(H(0), read_data=False) == chunk_short @@ -132,16 +134,48 @@ def test_read_data(repo_fixtures, request): def test_consistency(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: repository.put(H(0), fchunk(b"foo")) + repository.flush() # N>1: flush before reading the just-put chunk back assert pdchunk(repository.get(H(0))) == b"foo" repository.put(H(0), fchunk(b"foo2")) + repository.flush() assert pdchunk(repository.get(H(0))) == b"foo2" repository.put(H(0), fchunk(b"bar")) + repository.flush() assert pdchunk(repository.get(H(0))) == b"bar" # delete is a no-op for now (see Repository.delete): the latest put still wins. repository.delete(H(0)) assert pdchunk(repository.get(H(0))) == b"bar" +def test_multi_object_pack_roundtrip(repo_fixtures, request): + # Two objects fill one pack and must both read back: the second from a non-zero offset, and + # read_data=False returning only its header+meta. The test pins max_count=2 so it does not depend + # on the Repository.open() default; the compact tests override max_count via build_one_pack() too. + meta0, data0 = b"meta0", b"the-first-object" + meta1, data1 = b"m1", b"second" + chunk0 = fchunk(data0, meta=meta0, chunk_id=H(0)) + chunk1 = fchunk(data1, meta=meta1, chunk_id=H(1)) + with get_repository_from_fixture(repo_fixtures, request) as repository: + repository._pack_writer.max_count = 2 # this test is written for exactly two objects per pack + repository.put(H(0), chunk0) + assert repository.chunks[H(0)].pack_id == UNKNOWN_BYTES32 # buffered: the pack is not full yet + repository.put(H(1), chunk1) # fills the pack, flushing both objects at once + # both objects share one pack, written exactly once, laid out in put() order + pack_id = repository.chunks[H(0)].pack_id + assert pack_id != UNKNOWN_BYTES32 + assert repository.chunks[H(1)].pack_id == pack_id + assert [info.name for info in repository.store_list("packs")] == [bin_to_hex(pack_id)] + assert repository.chunks[H(0)].obj_offset == 0 + assert repository.chunks[H(1)].obj_offset == len(chunk0) # second object read from a non-zero offset + # full reads return each object's exact bytes, the second one resolved from its non-zero offset + assert repository.get(H(0)) == chunk0 + assert repository.get(H(1)) == chunk1 + # read_data=False returns header+meta only and stays inside the requested object + hdr_size = RepoObj.obj_header.size + assert repository.get(H(0), read_data=False) == chunk0[: hdr_size + len(meta0)] + assert repository.get(H(1), read_data=False) == chunk1[: hdr_size + len(meta1)] + + def build_one_pack(repository, objects): with repository: repository._pack_writer.max_count = len(objects) + 1 # prevent per-put flush; one pack on flush() @@ -226,6 +260,7 @@ def test_max_data_size(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: max_data = b"x" * (MAX_DATA_SIZE - RepoObj.obj_header.size) repository.put(H(0), fchunk(max_data)) + repository.flush() # N>1: make the buffered pack durable before get() assert pdchunk(repository.get(H(0))) == max_data with pytest.raises(IntegrityError): repository.put(H(1), fchunk(max_data + b"x")) @@ -399,13 +434,16 @@ def test_get_uses_chunk_index_location(tmp_path): def test_put_marks_id_in_chunk_index(tmp_path): - # put() immediately updates _chunks: add() marks the id as seen, then update_pack_info - # fills in the real pack location for the current session. + # At N>1, put() marks the id pending (pack_id=UNKNOWN_BYTES32); flush() then fills in the + # real pack location for the current session. with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository: id1 = H(1) repository.put(id1, fchunk(b"ZEROS")) entry = repository._chunks.get(id1) assert entry is not None + assert entry.pack_id == UNKNOWN_BYTES32 # buffered, not yet flushed + repository.flush() + entry = repository._chunks.get(id1) assert entry.pack_id == sha256(fchunk(b"ZEROS")).digest() assert entry.size == 0 # uncompressed size filled in by cache layer