From 5de314ba2ff8586a463dcad6b63099d0d77eee8a Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Mon, 6 Jul 2026 23:37:29 +0200 Subject: [PATCH] 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 --- src/borg/cache.py | 40 +++++++++++++------- src/borg/constants.py | 3 ++ src/borg/testsuite/cache_test.py | 65 ++++++++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 13 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index c73ac6280..a5bd75b9b 100644 --- a/src/borg/cache.py +++ b/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() diff --git a/src/borg/constants.py b/src/borg/constants.py index cd82509a8..8d4fa458c 100644 --- a/src/borg/constants.py +++ b/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 diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index f2a4bcf8d..3bdc39ff9 100644 --- a/src/borg/testsuite/cache_test.py +++ b/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.