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_INVALID_SENTINEL
 from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat
 from .helpers import get_cache_dir
 from .helpers import chunkit
@@ -551,6 +551,7 @@ def chunkindex_fragment_entry_size():
     return key_size + value_size
 
 
+# TODO: refactor the chunk-index functions below into a dedicated index-management class.
 def list_chunkindex_fragments(repository):
     """List the index/ fragments, returning each fragment's (name, approximate entry count).
 
@@ -576,14 +577,46 @@ def list_chunkindex_hashes(repository):
     return hashes
 
 
+def chunkindex_is_invalid(repository):
+    """Return whether the chunk-index-invalid marker is present.
+
+    store_list() bypasses the borgstore cache, so this stays correct even for a cache-backed namespace.
+    """
+    return any(ItemInfo(*info).name == CHUNKINDEX_INVALID_SENTINEL for info in repository.store_list("config"))
+
+
+def write_chunkindex_invalid(repository):
+    """Mark the chunk index as invalid. Call before deleting index fragments.
+
+    If the deletion is interrupted, the marker remains and the index is rebuilt on next load.
+    """
+    repository.store_store(f"config/{CHUNKINDEX_INVALID_SENTINEL}", b"")
+
+
+def delete_chunkindex_invalid(repository):
+    """Clear the chunk-index-invalid marker. Call after all fragment deletions have completed."""
+    try:
+        repository.store_delete(f"config/{CHUNKINDEX_INVALID_SENTINEL}")
+    except StoreObjectNotFound:
+        pass
+
+
 def delete_chunkindex_from_repo(repository):
     hashes = list_chunkindex_hashes(repository)
+    invalid = chunkindex_is_invalid(repository)
+    if hashes:
+        # mark invalid before deleting the first fragment, so an interrupted deletion is detectable.
+        write_chunkindex_invalid(repository)
     for hash in hashes:
         index_name = f"index/{hash}"
         try:
             repository.store_delete(index_name)
         except StoreObjectNotFound:
             pass
+    if hashes or invalid:
+        # clear the marker after every fragment is gone; also clears a marker left behind by an
+        # earlier interrupted deletion.
+        delete_chunkindex_invalid(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.
@@ -687,12 +720,21 @@ 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 invalid marker. 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_invalid(repository)
         for hash in delete_these:
             index_name = f"index/{hash}"
             try:
                 repository.store_delete(index_name)
             except StoreObjectNotFound:
                 pass
+        if guard:
+            delete_chunkindex_invalid(repository)
         if delete_these:
             logger.debug(f"chunk indexes deleted: {delete_these}")
     return new_hashes
@@ -729,6 +771,9 @@ 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.
     """
+    if chunkindex_is_invalid(repository):
+        # the index is invalid; it will be rebuilt on next load, so there is nothing to consolidate.
+        return
     small = [
         (name, approx)
         for name, approx in list_chunkindex_fragments(repository)
@@ -778,6 +823,15 @@ 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):
+            if chunkindex_is_invalid(repository):
+                # leftover 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 is invalid (interrupted operation), rebuilding it.")
+                try:
+                    delete_chunkindex_from_repo(repository)
+                except Exception as err:
+                    logger.debug(f"could not remove invalid chunk index fragments: {err!r}")
+                break
             hashes = list_chunkindex_hashes(repository)
             if not hashes:  # no chunk index fragments available
                 break

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
+# Marker object in the config/ namespace recording that the chunk index is invalid. Written before
+# deleting index fragments and removed once every fragment is gone. While present, the chunk index is
+# rebuilt from the packs on next load, so an interrupted fragment deletion cannot leave a partial
+# fragment set in use.
+CHUNKINDEX_INVALID_SENTINEL = "chunkindex-invalid"
 
 FD_MAX_AGE = 4 * 60  # 4 minutes
 

src/borg/repository.py

@@ -707,6 +707,10 @@ 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")
+        # a marker in config/ records that the chunk index is invalid (an interrupted fragment
+        # deletion); it will be rebuilt on next use, so warn rather than verify the leftover fragments.
+        if any(info.name == CHUNKINDEX_INVALID_SENTINEL for info in store_list("config")):
+            logger.warning("chunk index is invalid (interrupted operation); it will be rebuilt on next use.")
         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/cache_test.py

@@ -18,8 +18,10 @@
     list_chunkindex_hashes,
     read_chunkindex_from_repo,
     repack_chunkindex,
+    write_chunkindex_invalid,
     write_chunkindex_to_repo,
 )
+from ..constants import CHUNKINDEX_INVALID_SENTINEL
 from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..crypto.key import AESOCBKey
 from ..helpers import safe_ns
@@ -479,6 +481,74 @@ def test_repack_idempotent_deletes_sources_when_merged_exists(tmp_path, monkeypa
         assert set(merged) == set(all_keys)
 
 
+def _invalid_marker_present(repository):
+    return any(info.name == CHUNKINDEX_INVALID_SENTINEL for info in repository.store_list("config"))
+
+
+def test_invalid_marker_forces_rebuild(tmp_path):
+    """With the marker present, build discards leftover fragments instead of trusting them."""
+    loc = os.fspath(tmp_path / "repository")
+    with Repository(loc, exclusive=True, create=True) as repository:
+        # a leftover fragment plus the marker, as an interrupted deletion would leave them.
+        _seed_fragment(repository, 5000, 3)
+        write_chunkindex_invalid(repository)
+        assert _invalid_marker_present(repository)
+
+        chunks = build_chunkindex_from_repo(repository)
+        # the leftover entry is not merged (the repo has no packs, so a correct rebuild is empty).
+        assert _ci_key(5000) not in chunks
+        # the roll-forward removed both the leftover fragments and the marker.
+        assert not _invalid_marker_present(repository)
+        assert list_chunkindex_fragments(repository) == []
+
+
+def test_delete_chunkindex_writes_then_removes_marker(tmp_path):
+    """An interrupted delete_chunkindex_from_repo leaves the invalid marker behind."""
+    loc = os.fspath(tmp_path / "repository")
+    with Repository(loc, exclusive=True, create=True) as repository:
+        _seed_fragment(repository, 0, 3)
+        _seed_fragment(repository, 100, 3)
+        assert not _invalid_marker_present(repository)
+
+        real_delete = repository.store.delete
+        calls = {"n": 0}
+
+        def failing_delete(name, **kwargs):
+            # let the marker write happen, then fail on the first fragment deletion.
+            if name.startswith("index/"):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    raise OSError("simulated crash mid-delete")
+            return real_delete(name, **kwargs)
+
+        repository.store.delete = failing_delete
+        with pytest.raises(OSError):
+            delete_chunkindex_from_repo(repository)
+        repository.store.delete = real_delete
+        # the marker was written before the (failed) deletion and is still present.
+        assert _invalid_marker_present(repository)
+
+        # a fresh build sees the marker, rolls the deletion forward, and rebuilds cleanly.
+        build_chunkindex_from_repo(repository)
+        assert not _invalid_marker_present(repository)
+
+
+def test_repack_skips_when_invalid(tmp_path, monkeypatch):
+    """Repack does nothing while the index is invalid (a rebuild will supersede it)."""
+    monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000)
+    monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 2)
+    loc = os.fspath(tmp_path / "repository")
+    with Repository(loc, exclusive=True, create=True) as repository:
+        delete_chunkindex_from_repo(repository)
+        for j in range(4):
+            _seed_fragment(repository, j * 100, 100)
+        write_chunkindex_invalid(repository)
+        before = {name for name, _ in list_chunkindex_fragments(repository)}
+        repack_chunkindex(repository)
+        after = {name for name, _ in list_chunkindex_fragments(repository)}
+        assert before == after  # unchanged: repack bailed out
+
+
 def test_files_cache_save_tolerates_missing_chunk(tmp_path, monkeypatch):
     """A files-cache entry whose chunk vanished from the index is dropped, not fatal.
 

src/borg/testsuite/repository_test.py

@@ -969,6 +969,19 @@ def test_check_detects_index_corruption(tmp_path):
         assert repository.check(repair=False) is False  # mismatch between content hash and name detected
 
 
+def test_check_warns_on_invalid_chunk_index(tmp_path, caplog):
+    # a config/ marker records that the chunk index is invalid; check warns but does not fail,
+    # since the index is not part of the repository's object integrity.
+    import logging
+    from ..constants import CHUNKINDEX_INVALID_SENTINEL
+
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        repository.store_store(f"config/{CHUNKINDEX_INVALID_SENTINEL}", b"")
+        with caplog.at_level(logging.WARNING):
+            assert repository.check(repair=False) is True
+        assert "chunk index is invalid" in caplog.text
+
+
 def test_check_intact_multi_object_pack_passes(tmp_path):
     # An intact pack with several objects passes: it is hashed as a whole, so the object count
     # does not matter.