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

src/borg/cache.py

@@ -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 F_PENDING set: its pack location
+            # is not resolved yet. only serialize an entry once its pack is written (#9900).
+            assert not (entry.flags & ChunkIndex.F_PENDING), f"chunk {bin_to_hex(key)} has no pack location yet"
             # 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 they are still F_PENDING with no location (#9900).
+                self.repository.flush()
                 write_chunkindex_to_repo(self.repository, self._chunks, clear=clear)
             self.chunks_index_last_write = now
 
@@ -1004,8 +1011,6 @@ def open(self):
     def close(self):
         self.security_manager.save(self.manifest, self.key)
         pi = ProgressIndicatorMessage(msgid="cache.close")
-        if self._chunks is not None:
-            self.repository.flush()
         if self._files is not None:
             pi.output("Saving files cache")
             integrity_data = self._write_files_cache(self._files)

src/borg/testsuite/cache_test.py

@@ -1,5 +1,6 @@
 import os
 import time
+from datetime import UTC, datetime
 
 import pytest
 
@@ -66,6 +67,21 @@ 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, cache):
+        # #9900: the periodic index write must flush the pack writer first, so a buffered chunk is
+        # persisted with its real pack location rather than the unresolved F_PENDING placeholder.
+        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 matches
+        # the real (flushed) location, not the placeholder:
+        index = build_chunkindex_from_repo(repository)
+        assert H(7) in index
+        # the flush resolved H(7)'s location, so it is no longer pending/placeholder:
+        assert not repository.chunks.is_pending(H(7))
+        assert index[H(7)].pack_id == repository.chunks[H(7)].pack_id
+
     def test_files_cache(self, cache):
         st = os.stat(".")
         assert cache.file_known_and_unchanged(b"foo", bytes(32), st) == (False, None)