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 id -> 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,74 @@ def iter_headers(self):
             offset += obj_size
 
 
+class PackTracker:
+    """Packs check() verified in the current cycle, mapping pack_id (32 bytes) -> (timestamp, result).
+
+    A cycle is one full pass over packs/, which --max-duration may spread over several partial checks.
+    The set is stored at cache/checked-packs with a sha256 over the serialized table appended. load()
+    accepts the table only if that hash matches and its entries have the layout this class writes.
+    """
+
+    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 = self._new_table()
+
+    @classmethod
+    def _new_table(cls):
+        return HashTableNT(key_size=32, value_type=cls.Entry, value_format=cls._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:
+                table = HashTableNT.read(f)
+        except ValueError:
+            logger.warning("Ignoring unreadable checked-packs set.")
+            return
+        # read() rebuilds key size and value type from the blob, so a table written with a different
+        # Entry layout reads without error. All entries share one layout, so sampling one is enough.
+        for key, value in table.items():
+            if len(key) != 32 or value._fields != self.Entry._fields:
+                logger.warning("Ignoring checked-packs set with an unexpected layout.")
+                return
+            break
+        self.table = table
+
+    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):
+        self.table.clear()
+        try:
+            self.store.delete(self.NAME)
+        except StoreObjectNotFound:
+            pass
+
+
 class Repository:
     """borgstore-based key/value store."""
 
@@ -678,34 +749,23 @@ 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
         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, so 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()
         t_last_checkpoint = t_start
         index_files = index_errors = 0
         pack_files = pack_errors = 0
-        # check index and packs with separate progress indicators, each running from 0% to 100%.
-        # hash the index first, on full and partial checks alike: it is small, and a corrupt index
-        # already means the user must repair it (rebuilding the index re-reads all packs anyway), so we
-        # 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 and packs get separate progress indicators, each running from 0% to 100%.
+        # the index is checked first and in full, on partial checks too: it is small, and index errors
+        # stop the pack check below.
         index_infos = store_list("index")
         index_pi = ProgressIndicatorPercent(total=len(index_infos), msg="Checking index %3.0f%%", msgid="check.index")
         for info in index_infos:
@@ -718,37 +778,36 @@ def store_list(namespace):
             index_pi.show(current=len(index_infos))  # finish at 100%
         index_pi.finish()
         if index_errors == 0:
-            # list the packs only now: a corrupt index skips this entirely. packs are the bulk of the
-            # work and the part --max-duration splits.
+            # packs are the bulk of the work and the part --max-duration spreads over several checks.
             pack_infos = store_list("packs")
             pack_pi = ProgressIndicatorPercent(total=len(pack_infos), msg="Checking packs %3.0f%%", msgid="check.packs")
             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_pi.show(increase=1)  # advance for skipped packs too, so the bar tracks packs/, not work done
+                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):
-                    pack_errors += 1  # repair (salvage into a new pack, fix index) is not implemented yet
+                ok = verify("packs", info.name)
+                if not ok:
+                    pack_errors += 1
+                tracker.record(pack_id, ok)
                 now = time.monotonic()
-                if now > t_last_checkpoint + 300:  # checkpoint every 5 mins
+                # a checkpoint rewrites the whole table (41 bytes per pack), so keep the interval long.
+                if now > t_last_checkpoint + 30 * 60:
                     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

@@ -1,11 +1,16 @@
+import io
 import os
 import sys
+from collections import namedtuple
 from hashlib import sha256
 
 import pytest
+from borghash import HashTableNT
+
 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 +984,150 @@ 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 _spy_hash(repository, monkeypatch):
+    # collect the keys passed to store.hash. check() hashes a pack exactly when it verifies it.
+    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)
+    return hashed_keys
+
+
+def _store_intact_pack(repository):
+    intact = fchunk(b"INTACT", chunk_id=H(1))
+    intact_id = sha256(intact).digest()
+    pack_key = "packs/" + bin_to_hex(intact_id)
+    repository.store_store(pack_key, intact)
+    return intact_id, pack_key
+
+
+def test_check_partial_clears_recorded_corruption_when_intact(tmp_path, monkeypatch):
+    # a pack recorded corrupt is re-verified, and an intact result replaces the stale record.
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        intact_id, pack_key = _store_intact_pack(repository)
+
+        tracker = PackTracker(repository.store)
+        tracker.record(intact_id, ok=False)  # stale corrupt record
+        tracker.save()
+
+        hashed_keys = _spy_hash(repository, monkeypatch)
+
+        assert repository.check(repair=False, max_duration=3600) is True
+        assert pack_key in hashed_keys  # re-verified
+
+
+def test_check_partial_skips_pack_recorded_intact(tmp_path, monkeypatch):
+    # a pack recorded intact in this cycle is skipped when a partial check resumes.
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        intact_id, pack_key = _store_intact_pack(repository)
+
+        tracker = PackTracker(repository.store)
+        tracker.record(intact_id, ok=True)
+        tracker.save()
+
+        hashed_keys = _spy_hash(repository, monkeypatch)
+
+        assert repository.check(repair=False, max_duration=3600) is True
+        assert pack_key not in hashed_keys  # skipped, not re-verified
+
+
+def test_check_full_ignores_recorded_set(tmp_path, monkeypatch):
+    # a full check verifies every pack regardless of the recorded set, then drops the set.
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        intact_id, pack_key = _store_intact_pack(repository)
+
+        tracker = PackTracker(repository.store)
+        tracker.record(intact_id, ok=True)
+        tracker.save()
+
+        hashed_keys = _spy_hash(repository, monkeypatch)
+
+        assert repository.check(repair=False) is True
+        assert pack_key in hashed_keys  # verified
+
+        after = PackTracker(repository.store)
+        after.load()
+        assert len(after) == 0  # cycle complete, set dropped
+
+
+def test_check_checked_packs_ignores_foreign_entry_layout(tmp_path):
+    # load() drops a set whose entries have a different layout than Entry, even though its sha256 matches.
+    OtherEntry = namedtuple("OtherEntry", "timestamp result extra")
+    OtherFormat = namedtuple("OtherFormat", "timestamp result extra")
+    with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository:
+        table = HashTableNT(
+            key_size=32, value_type=OtherEntry, value_format=OtherFormat(timestamp="Q", result="B", extra="B")
+        )
+        table[H(1)] = OtherEntry(timestamp=123, result=1, extra=9)
+        with io.BytesIO() as f:
+            table.write(f)
+            data = f.getvalue()
+        repository.store_store(PackTracker.NAME, data + sha256(data).digest())
+
+        tracker = PackTracker(repository.store)
+        tracker.load()
+        assert len(tracker) == 0
+
+
 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