Merge pull request #10379 from mr-raj12/check-finish-narrow-walk-8466

check --repair: re-read only the packs the repair wrote, refs #8466

Author: ThomasWaldmann

docs/internals/data-structures.rst

@@ -86,9 +86,9 @@ cache/
     a marker object: while it is present, the chunks index in ``index/`` is considered
     invalid, because its fragments may be missing entries or point at deleted packs.
     It is written before deleting index fragments, before a single-object delete removes
-    the old pack, and before ``borg check --repair`` rebuilds the index after changing
-    the packs. It is removed after the last fragment is deleted or once the complete
-    current index is stored.
+    the old pack, and by ``borg check --repair`` after storing packs, before it re-reads
+    them and stores the index. It is removed after the last fragment is deleted or once
+    the complete current index is stored.
 
 Note that this ``cache/`` namespace is inside the repository (and thus shared by
 all clients); it is not the client-local cache described in

docs/internals/packs.rst

@@ -323,12 +323,12 @@ A deletion that could drop entries -- dropping the index entirely, or the full r
 above -- is guarded by a marker object, ``cache/chunkindex-invalid``, written before
 the first deletion and removed after the last one. A single-object delete writes the
 marker just before it removes the old pack, and ``borg check --repair`` writes it
-before rebuilding the index after changing the packs; both remove it once the index
-is stored. While the marker is present, the fragments may be missing entries or point
-at deleted packs, so they are not merged; the index is rebuilt from the pack files on
-the next load instead. A consolidation needs no marker: the entries of the small
-fragments it deletes are already contained in the merged fragments it wrote before
-deleting them.
+after storing packs, before it re-reads them and stores the index; both remove it
+once the index is stored. While the marker is present, the fragments may be missing
+entries or point at deleted packs, so they are not merged; the index is rebuilt from
+the pack files on the next load instead. A consolidation needs no marker: the entries
+of the small fragments it deletes are already contained in the merged fragments it
+wrote before deleting them.
 
 If the entire ``index/`` namespace is lost or corrupt, the ChunkIndex can be rebuilt
 by scanning pack files directly; see :ref:`pack-recovery`.

src/borg/archive.py

@@ -53,7 +53,8 @@
 from .patterns import PathPrefixPattern, FnmatchPattern, IECommand
 from .item import Item, ArchiveItem, ItemDiff
 from .platform import acl_get, acl_set, set_flags, get_flags, set_times, swidth
-from .repository import Repository
+from .hashindex import ChunkIndex, ChunkIndexEntry
+from .repository import Repository, PackReader
 from .repoobj import RepoObj, object_validator
 
 # macOS: SF_DATALESS marks dataless placeholder files (e.g. cloud files not materialized locally).
@@ -2236,9 +2237,27 @@ class ArchiveChecker:
     def __init__(self):
         self.error_found = False
         self.key = None
-        # True once repair drops a defect chunk or writes a new one, i.e. once the chunks index no
-        # longer matches the packs.
+        # True once repair changed the packs: it stored a chunk or deleted a defect chunk.
         self.chunks_modified = False
+        # ids of the packs repair wrote: stored by put() and flush(), or written by delete() rewriting a pack.
+        self.written_packs = set()
+
+    def record_stored(self, results):
+        """Add the pack ids in results to written_packs.
+
+        results: (chunk_id, pack_id, obj_offset, obj_size) tuples of the objects in the stored packs, as
+        Repository.put() and flush() return them, or None.
+        """
+        if results:
+            self.written_packs.update(pack_id for _, pack_id, _, _ in results)
+
+    def create_archive_entry(self, name, id, ts):
+        """Store the pack writer buffer, record the packs it wrote, create the archives directory entry.
+
+        Archives.create() stores the pack writer buffer too, but does not return the packs it wrote.
+        """
+        self.record_stored(self.repository.flush())
+        self.manifest.archives.create(name, id, ts)
 
     def note_dropped_objects(self):
         # The chunk index rebuild skipped repository content to get past a corrupt object header.
@@ -2295,7 +2314,8 @@ def check(
         self.repo_objs = RepoObj(self.key)
         validate = object_validator(self.repo_objs)
         # store the chunks buffered in the pack writer, so the index below has their pack locations
-        # (pack id, offset and size in the pack).
+        # (pack id, offset and size in the pack). The result is ignored: written_packs holds only the packs
+        # the repair writes.
         self.repository.flush()
         if not repair and self.repository.is_chunk_index_loaded:
             # without --repair, use the loaded index.
@@ -2423,9 +2443,14 @@ def verify_data(self):
                         # failed twice -> remove this defect chunk. delete rewrites its pack without it,
                         # keeping the other chunks, and removes it from self.chunks, so rebuild_archives
                         # reports the file it belongs to. update_index=False: finish() stores the index
-                        # rebuilt from the packs and clears the invalid marker delete() writes.
-                        self.repository.delete(defect_chunk, update_index=False, validate=validate)
+                        # and clears the invalid marker delete() writes.
+                        old_pack_id = self.chunks[defect_chunk].pack_id
+                        # new_pack_id: the pack holding the other objects of the old pack, None if there were none.
+                        new_pack_id, _ = self.repository.delete(defect_chunk, update_index=False, validate=validate)
                         self.chunks_modified = True
+                        self.written_packs.discard(old_pack_id)  # delete() removed the old pack
+                        if new_pack_id is not None:
+                            self.written_packs.add(new_pack_id)
                     else:
                         logger.warning("chunk %s not deleted, did not consistently fail.", bin_to_hex(defect_chunk))
             else:
@@ -2513,7 +2538,7 @@ def valid_archive(obj):
                     self.error_found = True
                     if self.repair:
                         logger.warning(f"Creating archives directory entry for {name} {archive_id_hex}.")
-                        self.manifest.archives.create(name, archive_id, archive.time)
+                        self.create_archive_entry(name, archive_id, archive.time)
                     else:
                         logger.warning(f"Would create archives directory entry for {name} {archive_id_hex}.")
 
@@ -2569,7 +2594,7 @@ def add_reference(id_, size, cdata):
             # with --repair, store a chunk the repository does not have; put() adds it to self.chunks.
             if self.repair and id_ not in self.chunks:
                 assert cdata is not None
-                self.repository.put(id_, cdata)
+                self.record_stored(self.repository.put(id_, cdata))
                 self.chunks_modified = True
 
         def verify_file_chunks(archive_name, item):
@@ -2782,51 +2807,125 @@ def valid_item(obj):
                     logger.debug(f"archive id new: {bin_to_hex(new_archive_id)}")
                     cdata = self.repo_objs.format(new_archive_id, {}, data, ro_type=ROBJ_ARCHIVE_META)
                     add_reference(new_archive_id, len(data), cdata)
-                    self.manifest.archives.create(info.name, new_archive_id, info.ts)
+                    self.create_archive_entry(info.name, new_archive_id, info.ts)
                     if archive_id != new_archive_id:
                         self.manifest.archives.delete_by_id(archive_id)
         finally:
             pi.finish()
             report_missing_chunks()
 
-    def finish(self):
-        if self.repair:
-            # flush chunks re-added during repair so their packs are on the store and out of the pack
-            # writer buffer (close() requires an empty buffer, #10055) before we (re)build the index.
-            self.repository.flush()
-            if self.chunks_modified:
-                # the packs changed: rebuild the index from them and store it. The index/ fragments lack
-                # the chunks this repair stored, so the index is invalid until the rebuilt one is stored.
-                # Free the current index first, so only one index is in memory.
-                write_chunkindex_invalid(self.repository)
-                self.repository.invalidate_chunk_index()
-                self.chunks = None
-                # Runs to completion, also after a Ctrl-C: delete_chunkindex_invalid() below declares
-                # the stored index to match the packs, which holds only once every pack was indexed.
-                if sig_int:
-                    logger.warning(
-                        "Rebuilding and writing the repository chunks index. "
-                        "This reads every pack and can not be interrupted."
+    def verify_written_packs(self):
+        """Read the object headers of the packs in written_packs and make the chunks index match them.
+
+        put() and delete() compute the index entries of the packs they write from the data they write.
+        This compares the (chunk_id, obj_offset, obj_size) of each object header in a written pack, read
+        with a validator, with the index entries that name the pack. Each difference sets error_found, is
+        logged and is fixed in the index:
+
+        - an index entry names an object the pack does not hold: the entry is removed.
+        - the pack holds an object whose chunk id is not indexed: the object is indexed.
+        - the pack does not exist: its index entries are removed.
+
+        A superseded duplicate is an object whose chunk id is indexed at another location. It is logged at
+        debug level. A pack delete() wrote holds one if compact_pack copied a byte range that no index entry
+        covers.
+        """
+        pack_ids = sorted(self.written_packs)
+        if not pack_ids:
+            return
+        logger.info(f"Re-reading {len(pack_ids)} pack(s) written by the repair.")
+        # (chunk_id, obj_offset, obj_size) of the index entries, per written pack.
+        indexed = {pack_id: set() for pack_id in pack_ids}
+        for chunk_id, entry in self.chunks.iteritems():
+            entries = indexed.get(entry.pack_id)
+            if entries is not None:
+                entries.add((chunk_id, entry.obj_offset, entry.obj_size))
+        validate = object_validator(self.repo_objs)
+        # pass 1 removes the index entries of every pack before pass 2 indexes any object, so whether an object
+        # is unindexed does not depend on the order the packs are read in.
+        found_in = {}  # pack_id -> (chunk_id, obj_offset, obj_size) of the objects in the pack
+        not_found_in = {}  # pack_id -> sorted index entries naming an object the pack does not hold
+        for pack_id in pack_ids:
+            # PackReader reads from the store, which does not refresh the repository lock.
+            self.repository._lock_refresh()
+            expected = indexed[pack_id]
+            key = "packs/" + bin_to_hex(pack_id)
+            info = self.repository.store.info(key)
+            if not info.exists:
+                self.error_found = True
+                logger.error(
+                    f"pack {bin_to_hex(pack_id)}: written by the repair, but it is missing. Removing its index entries."
+                )
+                for chunk_id, _, _ in expected:
+                    del self.chunks[chunk_id]
+                continue
+            reader = PackReader(self.repository.store, pack_id, pack_size=info.size)
+            found = list(reader.iter_headers(validate=validate, on_drop=self.note_dropped_objects))
+            not_found = sorted(expected.difference(found))
+            for chunk_id, _, _ in not_found:
+                del self.chunks[chunk_id]
+            found_in[pack_id] = found
+            not_found_in[pack_id] = not_found
+        # pass 2 indexes each unindexed object, so of several unindexed copies of a chunk, the first in pack id and
+        # offset order is indexed and the others are superseded duplicates.
+        for pack_id, found in found_in.items():
+            pack_hex = bin_to_hex(pack_id)
+            expected = indexed[pack_id]
+            not_found = not_found_in[pack_id]
+            unindexed = []
+            for obj in found:
+                chunk_id, obj_offset, obj_size = obj
+                if obj in expected:
+                    continue
+                if chunk_id in self.chunks:
+                    logger.debug(
+                        f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: "
+                        "superseded duplicate"
                     )
-                else:
-                    logger.info("Rebuilding and writing the repository chunks index.")
-                build_chunkindex_from_repo(
-                    self.repository,
-                    slow_rebuild=True,
-                    validate=object_validator(self.repo_objs),
-                    on_drop=self.note_dropped_objects,
-                    write_immediately=True,
+                    continue
+                unindexed.append(obj)
+                # size=0: the object header does not hold the plaintext size.
+                self.chunks[chunk_id] = ChunkIndexEntry(
+                    flags=ChunkIndex.F_USED, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size
                 )
-            else:
-                # the packs are unchanged, so the index still matches them: persist it as is.
-                logger.info("Writing the rebuilt repository chunks index.")
-                write_chunkindex_to_repo(
-                    self.repository, self.chunks, incremental=False, clear=False, force_write=True, delete_other=True
+            if not (not_found or unindexed):
+                continue
+            self.error_found = True
+            # an object the pack holds whose index entry has a wrong offset or size counts in both numbers,
+            # unless pass 2 indexed another copy of it first.
+            logger.error(
+                f"pack {pack_hex}: the chunks index does not match the pack. Indexed objects not in the pack: "
+                f"{len(not_found)}, objects in the pack with an unindexed chunk id: {len(unindexed)}. "
+                "Fixed the index."
+            )
+            for chunk_id, obj_offset, obj_size in not_found:
+                logger.debug(
+                    f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: not in pack"
+                )
+            for chunk_id, obj_offset, obj_size in unindexed:
+                logger.debug(
+                    f"pack {pack_hex}: {bin_to_hex(chunk_id)} at offset {obj_offset}, {obj_size} bytes: not indexed"
                 )
+
+    def finish(self):
+        if self.repair:
+            # store the pack writer buffer before the index is written (close() requires an empty buffer, #10055).
+            self.record_stored(self.repository.flush())
+            if self.chunks_modified or self.written_packs:
+                # the index/ fragments lack the pack changes of this repair.
+                write_chunkindex_invalid(self.repository)
+                # Runs to completion, also after a Ctrl-C: delete_chunkindex_invalid() below declares the
+                # stored index to match the packs, which holds only once every written pack was re-read.
+                self.verify_written_packs()
+            logger.info("Writing the repository chunks index.")
+            write_chunkindex_to_repo(
+                self.repository, self.chunks, incremental=False, clear=False, force_write=True, delete_other=True
+            )
+            # close() persists the in-memory index: drop it, the stored one is current.
+            self.repository.invalidate_chunk_index()
+            self.chunks = None
             # the stored index matches the packs: clear the invalid marker.
             delete_chunkindex_invalid(self.repository)
-            # drop the in-memory index so close() does not persist it over the index just written.
-            self.repository.invalidate_chunk_index()
 
 
 class ArchiveRecreater:

src/borg/archiver/check_cmd.py

@@ -222,9 +222,9 @@ def build_parser_check(self, subparsers, common_parser, mid_common_parser):
         ``borg check`` rebuilds the chunk index from the packs when ``--repair`` is given or when
         the stored index cannot be used. Ctrl-C stops that rebuild after the current object and
         discards the partial index: it lacks chunks that are still in the repository, so the check
-        would report them as lost. After a ``--repair`` that stored or deleted chunks, borg rebuilds and
-        stores the chunk index once more; that rebuild always runs to completion, also after a
-        Ctrl-C, and reads every pack.
+        would report them as lost. After a ``--repair`` that stored or deleted chunks, borg re-reads the
+        packs the repair wrote, makes the chunk index match them and stores it; that always runs to
+        completion, also after a Ctrl-C.
 
         About repair mode
         +++++++++++++++++

src/borg/cache.py

@@ -609,7 +609,7 @@ def write_chunkindex_invalid(repository):
     """Store the invalid marker, cache/chunkindex-invalid.
 
     Store it before deleting index/ fragments whose entries no other fragment holds, before deleting a pack
-    the fragments point at, and before rebuilding the index after pack changes the fragments do not record.
+    the fragments point at, and after storing packs the fragments do not record.
     While it is present, build_chunkindex_from_repo rebuilds the index from the packs instead of merging the
     fragments.
     """