diff --git a/docs/internals/data-structures.rst b/docs/internals/data-structures.rst index 84a80f6f36..6a91997613 100644 --- a/docs/internals/data-structures.rst +++ b/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/ diff --git a/src/borg/repository.py b/src/borg/repository.py index 580507519f..1c263be374 100644 --- a/src/borg/repository.py +++ b/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/ (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). diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index f9e4125979..c583857892 100644 --- a/src/borg/testsuite/repository_test.py +++ b/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