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

finish() validates the written packs against the shared index instead of rebuilding it from all packs.

Author: mr-raj12

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, NoManifestError
+from .hashindex import ChunkIndex, ChunkIndexEntry
+from .repository import Repository, NoManifestError, PackReader
 from .repoobj import RepoObj, object_validator
 
 # macOS: SF_DATALESS marks dataless placeholder files (e.g. cloud files not materialized locally).
@@ -2213,9 +2214,31 @@ 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 wrote a pack: it stored a chunk or deleted a defect chunk.
         self.chunks_modified = False
+        # True if check() built the chunks index from the packs with a validator, so every indexed object
+        # passed it. A validator (repoobj.object_validator) checks an object header against the object's
+        # tagged metadata.
+        self.rebuild_validated = False
+        # ids of the existing packs repair stored (put(), flush()) or wrote by rewriting a pack (delete()).
+        self.written_packs = set()
+
+    def record_stored(self, results):
+        """Add the pack ids in results to written_packs.
+
+        results: the (chunk_id, pack_id, obj_offset, obj_size) tuples Repository.put() or .flush() returns
+        for the packs it stored, or None if it stored no pack.
+        """
+        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, add the stored packs to written_packs, create the archives entry.
+
+        Archives.create() also stores the pack writer buffer, but does not return the stored packs.
+        """
+        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.
@@ -2305,6 +2328,7 @@ def check(
                 drop_corrupt_tail=repair,
                 write_immediately=False,
             )
+            self.rebuild_validated = repair and validate is not None
             # a rebuild from the packs sets F_NEW (entry not stored in the index/ fragments yet) on every
             # entry. Clear it, so Repository.close() does not store this index as a new fragment: only
             # finish() stores the index, and only with --repair.
@@ -2439,20 +2463,23 @@ def verify_data(self):
                             assert_id_place="verify_data",
                         )
                     except IntegrityErrorBase:
-                        # failed twice -> remove this defect chunk. delete rewrites its pack without it,
-                        # keeping the other chunks. update_index=False: finish() rebuilds the index from
-                        # the rewritten packs anyway, so a per-chunk full index write would be wasted.
-                        # delete() also removes the chunk from self.chunks, so rebuild_archives reports
-                        # the file it belongs to.
+                        # failed twice -> remove this defect chunk. delete() removes it from self.chunks,
+                        # so rebuild_archives reports the file it belongs to. update_index=False: finish()
+                        # writes the index.
                         if not index_marked_invalid:
                             # the index/ fragments point the other chunks of a rewritten pack at the
                             # deleted pack until finish() stores the new index. Mark them invalid before
                             # the first delete, so if the check stops before finish(), the next use
                             # rebuilds the index from the packs.
                             write_chunkindex_invalid(self.repository)
                             index_marked_invalid = True
-                        self.repository.delete(defect_chunk, update_index=False, validate=validate)
+                        old_pack_id = self.chunks[defect_chunk].pack_id
+                        new_pack_id, _ = self.repository.delete(defect_chunk, update_index=False, validate=validate)
                         self.chunks_modified = True
+                        # delete() replaced old_pack_id by new_pack_id, None if no object was left.
+                        self.written_packs.discard(old_pack_id)
+                        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:
@@ -2546,7 +2573,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}.")
 
@@ -2602,7 +2629,7 @@ def add_reference(id_, size, cdata):
             # --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):
@@ -2812,23 +2839,113 @@ 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 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 without reading the packs.
+        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 is a check finding,
+        logged and 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.
+
+        An object whose chunk id is indexed at another location is a superseded duplicate, not a finding.
+        A pack delete() wrote can hold one: compact_pack copies the byte ranges no index entry covers into
+        the new pack, except the superseded duplicates it finds there, and its search in such a range stops
+        at a corrupt object header.
+        """
+        pack_ids = sorted(self.written_packs)
+        if not pack_ids:
+            return
+        logger.info(f"Re-reading the packs written by the repair: {len(pack_ids)}.")
+        # (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)
+        for pack_id in pack_ids:
+            # PackReader reads from the store, which does not refresh the repository lock.
+            self.repository._lock_refresh()
+            pack_hex = bin_to_hex(pack_id)
+            expected = indexed.pop(pack_id)
+            reader = PackReader(self.repository.store, pack_id)
+            # iter_headers() yields nothing for a missing pack: the store reports size 0 for it.
+            if not self.repository.store.info(reader.key).exists:
+                self.error_found = True
+                logger.error(f"pack {pack_hex}: written by the repair, but it is missing. Removing its index entries.")
+                for chunk_id, _, _ in expected:
+                    del self.chunks[chunk_id]
+                continue
+            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]
+            # the loop indexes each unindexed object, so of several unindexed copies of a chunk, the first
+            # is indexed and the others are superseded duplicates.
+            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"
+                    )
+                    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
+                )
+            if not (not_found or unindexed):
+                continue
+            self.error_found = True
+            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:
-            # 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, validating every object header, and
-                # store it. Free the current index first, so only one is in memory.
-                self.repository.invalidate_chunk_index()
-                self.chunks = None
+            # store the pack writer buffer before the index is written (close() requires an empty buffer, #10055).
+            self.record_stored(self.repository.flush())
+            # without a validator (the key was not readable), check() indexed objects no validator
+            # checked, and skipped the rest of a pack after a corrupt object header. If repair wrote
+            # packs, rebuild the index from all packs with a validator.
+            full_rebuild = self.chunks_modified and not self.rebuild_validated
+            if not full_rebuild:
+                if self.chunks_modified:
+                    # the other packs did not change since check() indexed them with a validator.
+                    self.verify_written_packs()
+                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
+                )
+            # free the in-memory index: close() must not store it, and the full rebuild builds its own.
+            self.repository.invalidate_chunk_index()
+            self.chunks = None
+            if full_rebuild:
                 logger.info("Rebuilding and writing the repository chunks index.")
                 build_chunkindex_from_repo(
                     self.repository,
@@ -2837,17 +2954,9 @@ def finish(self):
                     on_drop=self.note_dropped_objects,
                     write_immediately=True,
                 )
-            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
-                )
             # the index just written matches the packs: clear the invalid marker, set by verify_data() or left
             # by an interrupted operation.
             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()
             self.manifest.write()
 
 

src/borg/repository.py

@@ -1115,10 +1115,15 @@ def is_chunk_index_loaded(self):
         return self._chunks is not None
 
     def flush(self):
-        """Flush any buffered pack writer chunks."""
+        """Store the pack writer buffer as a pack, after waiting for the pack the background store-thread is storing.
+
+        Returns the (chunk_id, pack_id, obj_offset, obj_size) tuples of the objects in the packs this call
+        stored or waited for, or None if there were none.
+        """
         if self._pack_writer is not None:
             self._lock_refresh()
-            self._pack_writer.flush()  # PackWriter updates _chunks internally
+            return self._pack_writer.flush()  # PackWriter updates _chunks internally
+        return None
 
     def close(self):
         if self._pack_writer is not None:
@@ -1173,11 +1178,10 @@ def check(self, repair=False, max_duration=0, max_age=0, repo_only=False):
         continuing. A read-only check never rebuilds the index: reading every pack to do so would be
         far too slow and expensive for a routine (e.g. cron) check. With repair=True and a corrupt
         index, and if every pack is intact, the index is rebuilt from the packs' object headers and
-        persisted; on a full check the archives phase rebuilds and re-persists it afterwards, see
-        ArchiveChecker.finish. Packs are verified by the store hash, which is content-addressing rather than a
-        MAC, so this rebuild detects accidental corruption but not tampering, refs #9901, #10026. If any
-        pack is corrupt the index is left unchanged, refs #8572, #10026. Pack ids found corrupt are kept
-        in cache/checked-packs, refs #9696.
+        persisted. Packs are verified by the store hash, which is content-addressing rather than a MAC,
+        so this rebuild detects accidental corruption but not tampering, refs #9901, #10026. If any pack
+        is corrupt the index is left unchanged, refs #8572, #10026. Pack ids found corrupt are kept in
+        cache/checked-packs, refs #9696.
 
         A pack recorded corrupt fails the check, also on a partial run that stops before re-reaching
         it. The record clears at the check that finds the pack intact again or gone (removed by
@@ -1573,11 +1577,12 @@ def put(self, id, data):
     def delete(self, id, *, validate, update_index=True):
         """Delete a single repo object by rewriting its pack without it (via compact_pack).
 
-        With update_index=True the full chunk index is written back so the next borg process sees the
-        deletion; callers that rebuild the index themselves (check --repair) pass update_index=False to
-        skip the per-object index rewrite.
-
         validate: passed to compact_pack.
+        update_index: if True, write the full chunk index to the repository after the delete, so the next
+            borg process sees it. If False, only the in-memory index is updated.
+
+        Returns compact_pack's (new_pack_id, dropped_bytes): the id of the pack holding the other objects
+        of the old pack (None if there were none), and the number of bytes the rewrite dropped.
         """
         self._lock_refresh()
         entry = self.chunks.get(id)
@@ -1587,13 +1592,14 @@ def delete(self, id, *, validate, update_index=True):
         # keep every object the chunk index lists for this pack, except the one being deleted.
         keep_ids = {cid for cid, e in self.chunks.iteritems() if e.pack_id == pack_id}
         keep_ids.discard(id)
-        self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id}, validate=validate)
+        result = self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id}, validate=validate)
         if update_index:
             # close() only persists new entries incrementally, so write the full index here to record
             # the removal for the next borg process.
             from .cache import write_chunkindex_to_repo
 
             write_chunkindex_to_repo(self, self.chunks, incremental=False, force_write=True, delete_other=True)
+        return result
 
     def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, validate, chunks=None):
         """Rewrite pack <pack_id>, keeping <keep_ids> and dropping <drop_ids>, then delete the old pack.