cache: sentinel to invalidate the chunk index during fragment deletion

src/borg/cache.py

@@ -22,7 +22,7 @@
 
 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_MERGE_ATTEMPTS
+from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_MERGE_ATTEMPTS, CHUNKINDEX_INVALIDATED_NAME
 from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat
 from .helpers import get_cache_dir
 from .helpers import chunkit
@@ -559,31 +559,63 @@ def list_chunkindex_fragments(repository):
     count is estimated from the stored object's byte size (chunkindex_fragment_entry_size() bytes per
     entry), so we can classify fragments (small vs. sealed) without loading them. The estimate ignores
     the small fixed header, which is negligible for the fragment sizes we care about.
-    Returns a list of (name, approx_entries) tuples, sorted by name.
+    Returns (fragments, invalidated): fragments is a list of (name, approx_entries) tuples sorted by
+    name; invalidated is True if the index-invalidated sentinel is present. The sentinel is excluded
+    from fragments. store_list() bypasses the borgstore cache, so the sentinel's presence stays
+    correct even when the index/ namespace is cache-backed.
     """
     entry_size = chunkindex_fragment_entry_size()
     fragments = []
+    invalidated = False
     for info in repository.store_list("index"):
         info = ItemInfo(*info)  # RPC does not give namedtuple
+        if info.name == CHUNKINDEX_INVALIDATED_NAME:
+            invalidated = True
+            continue
         fragments.append((info.name, info.size // entry_size))
     fragments.sort()
-    return fragments
+    return fragments, invalidated
 
 
 def list_chunkindex_hashes(repository):
-    hashes = [name for name, _ in list_chunkindex_fragments(repository)]  # already sorted by name
+    fragments, invalidated = list_chunkindex_fragments(repository)
+    hashes = [name for name, _ in fragments]  # already sorted by name
     logger.debug(f"chunk indexes: {hashes}")
-    return hashes
+    return hashes, invalidated
+
+
+def write_chunkindex_invalidated(repository):
+    """Write the index-invalidated sentinel. Call before deleting index fragments.
+
+    If the deletion is interrupted, the sentinel remains and the index is rebuilt on next load. Only
+    the sentinel's presence matters; its content is empty.
+    """
+    repository.store_store(f"index/{CHUNKINDEX_INVALIDATED_NAME}", b"")
+
+
+def delete_chunkindex_invalidated(repository):
+    """Remove the index-invalidated sentinel. Call after all fragment deletions have completed."""
+    try:
+        repository.store_delete(f"index/{CHUNKINDEX_INVALIDATED_NAME}")
+    except StoreObjectNotFound:
+        pass
 
 
 def delete_chunkindex_from_repo(repository):
-    hashes = list_chunkindex_hashes(repository)
+    hashes, invalidated = list_chunkindex_hashes(repository)
+    if hashes:
+        # mark invalidated before deleting the first fragment, so an interrupted deletion is detectable.
+        write_chunkindex_invalidated(repository)
     for hash in hashes:
         index_name = f"index/{hash}"
         try:
             repository.store_delete(index_name)
         except StoreObjectNotFound:
             pass
+    if hashes or invalidated:
+        # remove the sentinel after every fragment is gone; also clears a sentinel left behind by an
+        # earlier interrupted deletion.
+        delete_chunkindex_invalidated(repository)
     logger.debug(f"chunk indexes deleted: {hashes}")
     # the in-memory index is now stale; drop it so close() does not write it back into the
     # index we just deleted. the next .chunks access rebuilds it from actual repo contents.
@@ -626,7 +658,8 @@ def write_chunkindex_to_repo(
     #   bounded, immutable fragments over one big index.
     max_entries = CHUNKINDEX_FRAGMENT_ENTRIES_MAX
     # the fragment set present in the repo before we start writing:
-    stored_hashes = set(list_chunkindex_hashes(repository))
+    hashes, _ = list_chunkindex_hashes(repository)
+    stored_hashes = set(hashes)
     new_hashes = set()  # content hashes of the fragments that make up the index we are writing now
     fragments_written = 0
 
@@ -687,12 +720,20 @@ def write_chunkindex_to_repo(
             delete_these = set(stored_hashes) - new_hashes
         else:
             delete_these = set(delete_these) - new_hashes
+        # A delete_other rewrite may drop entries, so leftover fragments after a mid-delete crash must
+        # not be merged back; guard that deletion with the sentinel. Otherwise the deleted fragments'
+        # entries are already contained in the fragments we just wrote, so leftovers are harmless.
+        guard = delete_other and bool(delete_these)
+        if guard:
+            write_chunkindex_invalidated(repository)
         for hash in delete_these:
             index_name = f"index/{hash}"
             try:
                 repository.store_delete(index_name)
             except StoreObjectNotFound:
                 pass
+        if guard:
+            delete_chunkindex_invalidated(repository)
         if delete_these:
             logger.debug(f"chunk indexes deleted: {delete_these}")
     return new_hashes
@@ -729,11 +770,11 @@ def repack_chunkindex(repository):
     sum to >= MIN) or when too many small fragments have piled up (more than
     CHUNKINDEX_SMALL_FRAGMENT_CAP), so we don't rewrite a slowly growing fragment on every backup.
     """
-    small = [
-        (name, approx)
-        for name, approx in list_chunkindex_fragments(repository)
-        if approx < CHUNKINDEX_FRAGMENT_ENTRIES_MIN
-    ]
+    fragments, invalidated = list_chunkindex_fragments(repository)
+    if invalidated:
+        # the index is invalidated; it will be rebuilt on next load, so there is nothing to consolidate.
+        return
+    small = [(name, approx) for name, approx in fragments if approx < CHUNKINDEX_FRAGMENT_ENTRIES_MIN]
     if len(small) < 2:
         return  # nothing to gain from merging zero or one fragment
     small_total = sum(approx for _, approx in small)
@@ -778,7 +819,17 @@ def build_chunkindex_from_repo(
         # 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)
+            hashes, invalidated = list_chunkindex_hashes(repository)
+            if invalidated:
+                # the index is invalidated: surviving fragments may be incomplete or stale. Finish the
+                # interrupted deletion (best-effort; a read-only client rebuilds in memory only), then
+                # rebuild from packs.
+                logger.warning("chunk index was invalidated by an interrupted operation, rebuilding it.")
+                try:
+                    delete_chunkindex_from_repo(repository)
+                except Exception as err:
+                    logger.debug(f"could not remove invalidated chunk index fragments: {err!r}")
+                break
             if not hashes:  # no chunk index fragments available
                 break
             chunks = ChunkIndex()  # we'll merge all fragments into this

src/borg/constants.py

@@ -114,6 +114,11 @@
 # 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
+# Sentinel in the index/ namespace marking the chunk index as invalidated. Written before deleting
+# fragments and removed afterwards; while present, the index is rebuilt from packs. The all-zeros
+# name is a valid but unreachable sha256 (nothing hashes to it), so older borg reads it as an
+# unreadable fragment and falls back to the rebuild.
+CHUNKINDEX_INVALIDATED_NAME = "0" * 64
 
 FD_MAX_AGE = 4 * 60  # 4 minutes
 

src/borg/repository.py

@@ -707,6 +707,11 @@ def store_list(namespace):
         # stop and report that rather than continue. matters for partial checks too, whose runs can be
         # days apart (e.g. a weekend cron job).
         index_infos = store_list("index")
+        # the index-invalidated sentinel is not a real fragment and its content does not match its
+        # name; skip verification and warn rather than report it as corruption.
+        if any(info.name == CHUNKINDEX_INVALIDATED_NAME for info in index_infos):
+            logger.warning("chunk index was invalidated by an interrupted operation; it will be rebuilt on next use.")
+            index_infos = [info for info in index_infos if info.name != CHUNKINDEX_INVALIDATED_NAME]
         index_pi = ProgressIndicatorPercent(total=len(index_infos), msg="Checking index %3.0f%%", msgid="check.index")
         for info in index_infos:
             self._lock_refresh()

src/borg/testsuite/archiver/compact_cmd_test.py

@@ -139,7 +139,7 @@ def interrupt():
     # The objects were deleted, so no readable chunk index may still list them.
     repository = open_repository(archiver)
     with repository:
-        assert list_chunkindex_hashes(repository) == []
+        assert list_chunkindex_hashes(repository)[0] == []
 
     # A later backup of identical content must re-upload the deleted chunks, ...
     cmd(archiver, "create", "archive2", "input")
@@ -193,7 +193,7 @@ def store_delete_then_interrupt(name, **kwargs):
     # every persisted index entry points at a pack that still exists
     repository = open_repository(archiver)
     with repository:
-        assert list_chunkindex_hashes(repository) != []
+        assert list_chunkindex_hashes(repository)[0] != []
         pack_names_after = {info.name for info in repository.store_list("packs")}
         assert 0 < len(pack_names_after) < len(pack_names_before)  # some packs deleted, some left
         for id, entry in repository.chunks.iteritems():