check: track a set of checked packs for partial checks

docs/internals/data-structures.rst

@@ -31,12 +31,15 @@ config/
     the repository version encoded as decimal number text
   manifest
     some data about the repository, binary
-  last-pack-checked
-    repository check progress (partial checks, full checks' checkpointing),
-    key of last pack checked as text
   space-reserve.N
     purely random binary data to reserve space, e.g. for disk-full emergencies
 
+cache/
+  checked-packs
+    repository check progress (partial checks, full checks' checkpointing),
+    the set of packs checked so far this cycle (pack key -> timestamp, result),
+    as a hashtable with an appended integrity hash
+
 There is a list of pointers to archive objects in this directory:
 
 archives/

src/borg/repository.py

@@ -1,10 +1,13 @@
+import io
 import os
 import sys
 import time
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 from pathlib import Path
 from hashlib import sha256
 
+from borghash import HashTableNT
+
 from borgstore.store import Store
 from borgstore.backends.rest import REST, ssh_cmd
 from borgstore.store import ObjectNotFound as StoreObjectNotFound, ReadRangeError
@@ -53,7 +56,7 @@ def borg_permissions(permissions):
             return {
                 "": "lr",
                 "archives": "lrw",
-                "cache": "lrwWD",  # WD for last-pack-checked, ...
+                "cache": "lrwWD",  # WD for checked-packs, ...
                 "config": "lrW",  # W for manifest
                 "index": "lrwWD",  # WD for index/<HASH> (merge/compaction of incremental indexes)
                 "keys": "lr",
@@ -233,6 +236,60 @@ def iter_headers(self):
             offset += obj_size
 
 
+class PackTracker:
+    """Packs check() has verified in the current cycle, mapping pack_id (32 bytes) -> (timestamp, result).
+
+    is_intact() drives the skip in a partial check: intact packs are skipped, packs recorded corrupt are
+    re-verified. save() appends a sha256 over the serialized table; load() drops a blob whose sha256 does
+    not match, so a rotted set re-checks everything instead of skipping an unverified pack.
+    """
+
+    NAME = "cache/checked-packs"
+    Entry = namedtuple("Entry", "timestamp result")
+    EntryFormatT = namedtuple("EntryFormatT", "timestamp result")
+    _EntryFormat = EntryFormatT(timestamp="Q", result="B")  # unix ts, 1=ok 0=corrupt
+
+    def __init__(self, store):
+        self.store = store
+        self.table = HashTableNT(key_size=32, value_type=self.Entry, value_format=self._EntryFormat)
+
+    def __len__(self):
+        return len(self.table)
+
+    def is_intact(self, pack_id):
+        entry = self.table.get(pack_id)
+        return entry is not None and bool(entry.result)
+
+    def record(self, pack_id, ok):
+        self.table[pack_id] = self.Entry(timestamp=int(time.time()), result=int(ok))
+
+    def load(self):
+        try:
+            data = self.store.load(self.NAME)
+        except StoreObjectNotFound:
+            return
+        if len(data) < 32 or sha256(data[:-32]).digest() != data[-32:]:
+            logger.warning("Ignoring corrupted checked-packs set.")
+            return
+        try:
+            with io.BytesIO(data[:-32]) as f:
+                self.table = HashTableNT.read(f)
+        except ValueError:
+            logger.warning("Ignoring unreadable checked-packs set.")
+
+    def save(self):
+        with io.BytesIO() as f:
+            self.table.write(f)
+            data = f.getvalue()
+        self.store.store(self.NAME, data + sha256(data).digest())
+
+    def clear(self):
+        try:
+            self.store.delete(self.NAME)
+        except StoreObjectNotFound:
+            pass
+
+
 class Repository:
     """borgstore-based key/value store."""
 
@@ -678,23 +735,14 @@ def store_list(namespace):
         partial = bool(max_duration)
         assert not (repair and partial)
         mode = "partial" if partial else "full"
-        LAST_PACK_CHECKED = "cache/last-pack-checked"
         logger.info(f"Starting {mode} repository check")
+        tracker = PackTracker(self.store)
         if partial:
-            # continue a past partial check (if any) or from a checkpoint or start one from beginning
-            try:
-                last_pack_checked = self.store.load(LAST_PACK_CHECKED).decode()
-            except StoreObjectNotFound:
-                last_pack_checked = ""
+            tracker.load()  # resume the cycle: packs already recorded intact are skipped below.
         else:
-            # start from the beginning and also forget about any potential past partial checks
-            last_pack_checked = ""
-            try:
-                self.store.delete(LAST_PACK_CHECKED)
-            except StoreObjectNotFound:
-                pass
-        if last_pack_checked:
-            logger.info(f"Skipping to packs after {last_pack_checked}.")
+            tracker.clear()  # a full check verifies every pack; start a new cycle.
+        if len(tracker):
+            logger.info(f"Continuing check cycle, {len(tracker)} packs already checked.")
         else:
             logger.info("Starting from beginning.")
         t_start = time.monotonic()
@@ -725,30 +773,29 @@ def store_list(namespace):
             for info in pack_infos:
                 self._lock_refresh()
                 pack_pi.show(increase=1)  # advance for every pack, including ones a partial resume skips below
-                key = "packs/%s" % info.name
-                if key <= last_pack_checked:  # needs sorted keys
+                pack_id = hex_to_bin(info.name)
+                if tracker.is_intact(pack_id):  # verified intact earlier in this cycle
                     continue
                 pack_files += 1
-                if not verify("packs", info.name):
+                ok = verify("packs", info.name)
+                if not ok:
                     pack_errors += 1  # repair (salvage into a new pack, fix index) is not implemented yet
+                tracker.record(pack_id, ok)
                 now = time.monotonic()
-                if now > t_last_checkpoint + 300:  # checkpoint every 5 mins
+                if now > t_last_checkpoint + 30 * 60:  # checkpoint every 30 minutes
                     t_last_checkpoint = now
-                    logger.info(f"Checkpointing at pack {key}.")
-                    self.store.store(LAST_PACK_CHECKED, key.encode())
+                    logger.info(f"Checkpointing at pack {info.name}.")
+                    tracker.save()
                 if partial and now > t_start + max_duration:
-                    logger.info(f"Finished partial repository check, last pack checked is {key}.")
-                    self.store.store(LAST_PACK_CHECKED, key.encode())
+                    logger.info(f"Finished partial repository check, {len(tracker)} packs checked so far.")
+                    tracker.save()
                     break
             else:
-                # the pack scan reached the end (no partial timeout): the check is complete, drop the checkpoint.
+                # scanned all packs without hitting the time limit: the cycle is done, drop the set.
                 if pack_infos:
                     pack_pi.show(current=len(pack_infos))  # finish at 100%
                 logger.info("Finished checking packs.")
-                try:
-                    self.store.delete(LAST_PACK_CHECKED)
-                except StoreObjectNotFound:
-                    pass
+                tracker.clear()
             pack_pi.finish()
         else:
             # TODO: --repair will rebuild the index from the packs here instead of stopping (refs #8572).

src/borg/testsuite/repository_test.py

@@ -6,6 +6,7 @@
 from ..helpers import IntegrityError, Location, bin_to_hex
 from ..hashindex import ChunkIndex
 from ..repository import Repository, MAX_DATA_SIZE, rest_serve_command, PackWriter, PackReader
+from ..repository import PackTracker
 from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION
 from .hashindex_test import H
 
@@ -979,6 +980,87 @@ def test_check_intact_multi_object_pack_passes(tmp_path):
         assert repository.check(repair=False) is True
 
 
+def test_check_checked_packs_roundtrip(tmp_path):
+    # the set survives a store/load round-trip; a rotted blob loads as empty.
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        tracker = PackTracker(repository.store)
+        tracker.table[H(1)] = PackTracker.Entry(timestamp=123, result=1)
+        tracker.table[H(2)] = PackTracker.Entry(timestamp=456, result=0)
+        tracker.save()
+
+        loaded = PackTracker(repository.store)
+        loaded.load()
+        assert len(loaded) == 2
+        assert H(1) in loaded.table and H(2) in loaded.table
+        assert tuple(loaded.table[H(2)]) == (456, 0)
+
+        corrupted = bytearray(repository.store.load(PackTracker.NAME))
+        corrupted[0] ^= 0xFF  # break the appended sha256
+        repository.store.store(PackTracker.NAME, bytes(corrupted))
+        rotted = PackTracker(repository.store)
+        rotted.load()
+        assert len(rotted) == 0
+
+
+def test_check_partial_rechecks_pack_sorting_before_checked_one(tmp_path):
+    # a partial check verifies a new pack even when its id sorts before an already-checked pack.
+    intact = fchunk(b"INTACT", chunk_id=H(1))
+    intact_id = sha256(intact).digest()
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        repository.store_store("packs/" + bin_to_hex(intact_id), intact)
+
+        # mark the intact pack as already checked in this cycle.
+        tracker = PackTracker(repository.store)
+        tracker.record(intact_id, ok=True)
+        tracker.save()
+
+        # add a corrupt pack (content does not hash to its name) whose id sorts before intact_id.
+        early_id = b"\x00" * 32
+        assert bin_to_hex(early_id) < bin_to_hex(intact_id)
+        repository.store_store("packs/" + bin_to_hex(early_id), b"CORRUPT-does-not-match-name")
+
+        assert repository.check(repair=False, max_duration=3600) is False
+
+
+def test_check_partial_rechecks_pack_recorded_corrupt(tmp_path):
+    # a pack recorded corrupt earlier in the cycle is re-verified, so the corruption keeps being reported.
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        corrupt_id = H(1)  # stored content does not hash to this name
+        repository.store_store("packs/" + bin_to_hex(corrupt_id), b"CORRUPT-does-not-match-name")
+
+        tracker = PackTracker(repository.store)
+        tracker.record(corrupt_id, ok=False)
+        tracker.save()
+
+        assert repository.check(repair=False, max_duration=3600) is False
+
+
+def test_check_partial_clears_recorded_corruption_when_intact(tmp_path, monkeypatch):
+    # a pack recorded corrupt is re-verified, not skipped: assert True alone would also hold if the
+    # pack were skipped outright, so this spies on store.hash to confirm verify() actually ran on it.
+    intact = fchunk(b"INTACT", chunk_id=H(1))
+    intact_id = sha256(intact).digest()
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        pack_key = "packs/" + bin_to_hex(intact_id)
+        repository.store_store(pack_key, intact)
+
+        tracker = PackTracker(repository.store)
+        tracker.record(intact_id, ok=False)  # stale corrupt record
+        tracker.save()
+
+        hashed_keys = []
+        orig_hash = repository.store.hash
+
+        def spy_hash(key):
+            hashed_keys.append(key)
+            return orig_hash(key)
+
+        monkeypatch.setattr(repository.store, "hash", spy_hash)
+
+        assert repository.check(repair=False, max_duration=3600) is True
+        assert pack_key in hashed_keys  # verify() ran on the pack; it was not skipped
+
+
 def test_check_progress_covers_packs_and_index(tmp_path, monkeypatch):
     # check() uses a separate progress indicator for index/ and for packs/. Each one is sized to its own
     # namespace and driven to 100% by a final show(current=total). A fake indicator records the wiring