cache: never build a partially merged chunk index, retry on vanished fragment

A concurrent repack_chunkindex (another client closing its cache, under a
shared lock) stores a merged replacement fragment and then deletes the small
source fragments. A reader that listed index/* before the replacement existed
and loaded the fragments after the deletion would silently skip the vanished
fragment and return a partially merged chunk index: chunks that exist in the
repo become invisible (spurious ObjectNotFound on read, lost deduplication,
duplicate chunk copies written into new packs).

Make the fragment merge all-or-nothing: if any listed fragment fails to load,
discard the partial merge, re-list and retry - the fresh listing contains the
replacement fragment. If no complete, consistent fragment set can be read
within CHUNKINDEX_MERGE_ATTEMPTS, fall back to the slow (but correct) index
rebuild from the pack headers instead of returning an incomplete index.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE
+from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE, CHUNKINDEX_MERGE_ATTEMPTS
 from .hashindex import ChunkIndex, ChunkIndexEntry
 from .helpers import get_cache_dir
 from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list
@@ -763,20 +763,31 @@ def build_chunkindex_from_repo(
 ):
     # first, try to build a fresh, mostly complete chunk index from centrally cached chunk indexes:
     if not disable_caches:
-        hashes = list_chunkindex_hashes(repository)
-        if hashes:  # we have at least one cached chunk index!
-            merged = 0
-            chunks = ChunkIndex()  # we'll merge all we find into this
+        # a concurrent repack_chunkindex (another client, shared lock) deletes the small fragments it
+        # merged - only after storing the merged replacement, so no entries are ever lost, but a
+        # fragment we just listed can vanish before we load it. the index must be built from ALL
+        # fragments or not at all: a partially merged index would miss chunks that exist in the repo
+        # (spurious ObjectNotFound, lost deduplication). so on a failed load, re-list and retry -
+        # the fresh listing contains the replacement fragment. if we cannot get a complete, consistent
+        # set (e.g. a persistently unreadable fragment), fall through to the slow rebuild instead.
+        for _ in range(CHUNKINDEX_MERGE_ATTEMPTS):
+            hashes = list_chunkindex_hashes(repository)
+            if not hashes:  # no cached chunk index available
+                break
+            chunks = ChunkIndex()  # we'll merge all fragments into this
+            complete = True
             for hash in hashes:
                 chunks_to_merge = read_chunkindex_from_repo(repository, hash)
-                if chunks_to_merge is not None:
-                    logger.debug(f"cached chunk index {hash} gets merged...")
-                    for k, v in chunks_to_merge.items():
-                        chunks[k] = v
-                    merged += 1
-                    chunks_to_merge.clear()
-            if merged > 0:
-                if merged > 1 and cache_immediately:
+                if chunks_to_merge is None:
+                    logger.debug(f"cached chunk index {hash} vanished or is invalid, restarting the merge...")
+                    complete = False
+                    break
+                logger.debug(f"cached chunk index {hash} gets merged...")
+                for k, v in chunks_to_merge.items():
+                    chunks[k] = v
+                chunks_to_merge.clear()
+            if complete:
+                if len(hashes) > 1 and cache_immediately:
                     # consolidate small fragments on the repo so they don't pile up. this is a
                     # bounded repack: it does not collapse large, already-sealed fragments, so those
                     # stay immutable (and cache-stable for other clients) rather than being re-uploaded.
@@ -785,6 +796,9 @@ def build_chunkindex_from_repo(
                 # already holds these entries in its fragments.
                 chunks.clear_new()
                 return chunks
+            chunks.clear()  # free the partial merge before retrying
+        else:
+            logger.warning("could not read a complete set of cached chunk index fragments, rebuilding the index.")
     # if we didn't get anything from the cache, compute the ChunkIndex the slow way:
     logger.debug("rebuilding the chunk index from the repo the slow way...")
     chunks = ChunkIndex()

src/borg/constants.py

@@ -107,6 +107,9 @@ CHUNKINDEX_SMALL_FRAGMENT_CAP = 15
 # + obj_offset 4 + obj_size 4). Only used to estimate a fragment's entry count from its byte size,
 # so we can classify fragments without loading them.
 CHUNKINDEX_ENTRY_SIZE = 80
+# How often to restart merging the fragments into a chunk index when a listed fragment vanishes
+# mid-merge (a concurrent repack replaced it). After that, fall back to the slow rebuild from packs.
+CHUNKINDEX_MERGE_ATTEMPTS = 3
 
 FD_MAX_AGE = 4 * 60  # 4 minutes
 

src/borg/testsuite/cache_test.py

@@ -127,6 +127,71 @@ def test_read_chunkindex_from_repo_missing(tmp_path):
         assert result is None
 
 
+def test_build_chunkindex_retries_on_vanished_fragment(tmp_path):
+    """build_chunkindex_from_repo must re-list and retry when a fragment vanishes mid-merge.
+
+    A concurrent repack_chunkindex (another client closing its cache, shared lock) stores a merged
+    replacement fragment and then deletes the small source fragments. A reader that listed the
+    fragments before the replacement existed and loads them after the deletion must not silently
+    skip the vanished fragment (that would yield an incomplete index: spurious ObjectNotFound for
+    existing chunks, lost deduplication) - it must restart from a fresh listing.
+    """
+    from borgstore.store import ObjectNotFound as StoreObjectNotFound
+
+    repository_location = os.fspath(tmp_path / "repository")
+    with Repository(repository_location, exclusive=True, create=True) as repository:
+        # two fragments, as left behind by two incremental backups
+        for h in (H(1), H(2)):
+            ci = ChunkIndex()
+            ci[h] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, h, 0, 4)
+            write_chunkindex_to_repo(repository, ci, incremental=False, force_write=True)
+        fragment_names = list_chunkindex_hashes(repository)
+        original_store_load = repository.store_load
+        raced = False
+
+        def racing_store_load(name, **kwargs):
+            nonlocal raced
+            if not raced and name.startswith("index/"):
+                raced = True
+                # simulate a concurrent repack: store the merged replacement fragment first,
+                # then delete the sources, then fail this load (the fragment has vanished).
+                merged = ChunkIndex()
+                for h in (H(1), H(2)):
+                    merged[h] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, h, 0, 4)
+                write_chunkindex_to_repo(repository, merged, incremental=False, force_write=True)
+                for frag in fragment_names:
+                    repository.store_delete(f"index/{frag}")
+                raise StoreObjectNotFound(name)
+            return original_store_load(name, **kwargs)
+
+        repository.store_load = racing_store_load
+        try:
+            chunks = build_chunkindex_from_repo(repository)
+        finally:
+            repository.store_load = original_store_load
+        assert raced  # the simulated repack fired
+        # the retry picked up the replacement fragment: the index is complete
+        assert H(1) in chunks
+        assert H(2) in chunks
+
+
+def test_build_chunkindex_no_partial_merge(tmp_path):
+    """A persistently unreadable fragment must cause a rebuild from the packs, never a partial index."""
+    repository_location = os.fspath(tmp_path / "repository")
+    with Repository(repository_location, exclusive=True, create=True) as repository:
+        # a valid fragment ...
+        ci = ChunkIndex()
+        ci[H(1)] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, H(1), 0, 4)
+        write_chunkindex_to_repo(repository, ci, incremental=False, force_write=True)
+        # ... and a corrupt one (content does not match its name), which never loads successfully
+        repository.store_store(f"index/{'e' * 64}", b"garbage")
+        chunks = build_chunkindex_from_repo(repository)
+        # merging must have been abandoned in favor of the slow rebuild from the (empty) packs
+        # namespace; with the old skip-and-continue behavior, H(1) would be present here.
+        assert H(1) not in chunks
+        assert len(chunks) == 0
+
+
 def test_chunkindex_cache_not_consolidated_on_access(tmp_path):
     """ChunksMixin.chunks binds the repository index without collapsing the cached fragments.