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

@@ -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)

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"")
 

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

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
 
 

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):

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