cache: flush pack writer before periodic chunk-index write (#9900)

src/borg/cache.py

@@ -21,7 +21,7 @@
 from borgstore.store import ItemInfo
 
 from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS
-from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX
+from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX, UNKNOWN_BYTES32
 from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_MERGE_ATTEMPTS
 from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat
 from .helpers import get_cache_dir
@@ -653,8 +653,12 @@ def write_chunkindex_to_repo(
         # pre-size the temporary table to this batch so filling it does not repeatedly rehash:
         batch = ChunkIndex(usable=len(batch_keys) or None)
         for key in batch_keys:
+            entry = chunks[key]
+            # a chunk still buffered in the pack writer has UNKNOWN_BYTES32 as its pack_id;
+            # only serialize an entry once its pack is written and its pack_id is real (#9900).
+            assert entry.pack_id != UNKNOWN_BYTES32, f"chunk {bin_to_hex(key)} has no pack location"
             # for now, we don't want to serialize the flags or the size:
-            batch[key] = chunks[key]._replace(flags=ChunkIndex.F_NONE, size=0)
+            batch[key] = entry._replace(flags=ChunkIndex.F_NONE, size=0)
         new_hash, stored = _store_chunkindex_fragment(repository, batch, stored_hashes, force_write=force_write)
         batch.clear()  # free memory of the temporary table
         new_hashes.add(new_hash)
@@ -913,6 +917,9 @@ def add_chunk(
     def _maybe_write_chunks_index(self, now, force=False, clear=False):
         if force or now > self.chunks_index_last_write + self.chunks_index_write_td:
             if self._chunks is not None:
+                # flush the pack writer first, so buffered chunks get their real pack location;
+                # until their pack is written their pack_id is the UNKNOWN_BYTES32 placeholder (#9900).
+                self.repository.flush()
                 write_chunkindex_to_repo(self.repository, self._chunks, clear=clear)
             self.chunks_index_last_write = now
 

src/borg/testsuite/cache_test.py

@@ -1,5 +1,6 @@
 import os
 import time
+from datetime import UTC, datetime
 
 import pytest
 
@@ -19,6 +20,7 @@
     repack_chunkindex,
     write_chunkindex_to_repo,
 )
+from ..constants import UNKNOWN_BYTES32
 from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..crypto.key import AESOCBKey
 from ..helpers import safe_ns
@@ -66,6 +68,19 @@ def test_existing_reuse_after_add_chunk(self, cache):
         assert cache.add_chunk(H(1), {}, b"5678", stats=Statistics()) == (H(1), 4)
         assert cache.reuse_chunk(H(1), 4, Statistics()) == (H(1), 4)
 
+    def test_periodic_index_write_flushes_pending(self, repository, key, manifest):
+        # #9900: the periodic index write must flush the pack writer first, so a buffered chunk is
+        # persisted with its real pack location instead of the UNKNOWN_BYTES32 placeholder.
+        cache = AdHocWithFilesCache(manifest)
+        cache.add_chunk(H(7), {}, b"h1-payload", stats=Statistics())
+        assert repository.chunks.is_pending(H(7))  # H(7) is buffered, not yet in a pack
+        # force=True runs the same write the 600s timer triggers:
+        cache._maybe_write_chunks_index(datetime.now(UTC), force=True)
+        # rebuild the index from the persisted fragments and check H(7)'s stored location:
+        index = build_chunkindex_from_repo(repository)
+        assert H(7) in index
+        assert index[H(7)].pack_id != UNKNOWN_BYTES32
+
     def test_files_cache(self, cache):
         st = os.stat(".")
         assert cache.file_known_and_unchanged(b"foo", bytes(32), st) == (False, None)