From 23afa5dccf6758c1dffaca20a00be4d08ec1a52c Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Fri, 3 Jul 2026 00:12:58 +0200 Subject: [PATCH 1/9] Repack chunks index into entry-count-bounded fragments Replace the unconditional "collapse all index/* fragments into one all-in-one index" behavior with an entry-count-based repacking policy that keeps each fragment within [CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX] entries where possible. - write_chunkindex_to_repo now always splits its output into fragments of at most MAX entries (streamed, so memory stays bounded), collecting all new hashes so delete_other/delete_these never delete a fragment just written. - repack_chunkindex() merges small (< MIN) fragments, deferring until they can seal a full fragment (sum >= MIN) or too many piled up (> SMALL_FRAGMENT_CAP); fragments already in range are left untouched and immutable. - Wire repack into AdHocWithFilesCache.close() and the build_chunkindex_from_repo merge>1 branch (which no longer collapses-and-deletes everything). - list_chunkindex_fragments() estimates a fragment's entry count from its stored byte size, so fragments can be classified without loading them. This bounds both fragment size and count, and keeps large fragments stable (groundwork for a future writethrough cache on the index/ namespace, which an all-in-one consolidation would invalidate for every client of the repo). Also: Make chunk index fragment partitioning deterministic Sort the selected keys before partitioning them into fragments, so an identical set of entries always produces an identical fragment set (identical content hashes), no matter in which order the entries were inserted into the hash table. This makes writing/repacking idempotent and convergent across clients: a fragment that already exists in the repo is not stored again, and no differently-partitioned duplicates of the same entries can pile up (previously, two clients repacking the same small fragments could each produce a different - possibly sealed, thus never cleaned up - merged fragment). Sorting cost is negligible on the hot paths: incremental writes and repacks sort ~100k-200k keys (~0.02s). Only full rewrites (compact, slow rebuild) sort the whole index, where the sort is dwarfed by the rest of the operation. Also: Don't fail an already-committed backup on repack error The repack in ChunksMixin.close() runs after the archive and its incremental chunk-index fragment are durably stored, so it is optional maintenance. Catch a transient repack failure and warn instead of letting it propagate out of close() and fail the whole command; the next run repacks again. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 222 +++++++++++++++++++++++------- src/borg/constants.py | 13 ++ src/borg/testsuite/cache_test.py | 225 ++++++++++++++++++++++++++++++- 3 files changed, 407 insertions(+), 53 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index 4855baa57a..75d336e05d 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -20,6 +20,8 @@ from borgstore.store import ItemInfo from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS +from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX +from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE from .hashindex import ChunkIndex, ChunkIndexEntry from .helpers import get_cache_dir from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list @@ -525,6 +527,22 @@ def list_chunkindex_hashes(repository): return hashes +def list_chunkindex_fragments(repository): + """Like list_chunkindex_hashes, but also return each fragment's approximate entry count. + + The entry count is estimated from the stored object's byte size (CHUNKINDEX_ENTRY_SIZE bytes + per entry), so we can classify fragments (small vs. sealed) without loading them. The estimate + ignores the small fixed header, which is negligible for the fragment sizes we care about. + Returns a list of (name, approx_entries) tuples, sorted by name. + """ + fragments = [] + for info in repository.store_list("index"): + info = ItemInfo(*info) # RPC does not give namedtuple + fragments.append((info.name, info.size // CHUNKINDEX_ENTRY_SIZE)) + fragments.sort() + return fragments + + def delete_chunkindex_from_repo(repository): hashes = list_chunkindex_hashes(repository) for hash in hashes: @@ -539,58 +557,104 @@ def delete_chunkindex_from_repo(repository): repository.invalidate_chunk_index() -def write_chunkindex_to_repo( - repository, chunks, *, incremental=True, clear=False, force_write=False, delete_other=False, delete_these=None -): - # for now, we don't want to serialize the flags or the size: - chunks_to_write = ChunkIndex() - # incremental==True: - # the borghash code has no means to only serialize the F_NEW table entries, - # thus we copy only the new entries to a temporary table. - # incremental==False: - # maybe copying the stuff into a new ChunkIndex is not needed here, - # but for simplicity, we do it anyway. - for key, existing in chunks.iteritems(only_new=incremental): - chunks_to_write[key] = existing._replace(flags=ChunkIndex.F_NONE, size=0) - num_to_write = len(chunks_to_write) +def _store_chunkindex_fragment(repository, batch, stored_hashes, *, force_write): + """Serialize a temporary ChunkIndex `batch` and store it as an index/ fragment. + + We don't serialize the flags or the size, so callers pass entries with those zeroed. The object + is stored under index/, where is the sha256 of its content, so borgstore can verify + it like any other object; an incompatible format from a different borg version is rejected by + borghash's own versioned header (MAGIC + VERSION) when read back. + + Returns (new_hash, stored) where `stored` is True iff we actually wrote to the repository (we skip + the write if a fragment with the same content hash already exists and force_write is not set). + """ with io.BytesIO() as f: - chunks_to_write.write(f) + batch.write(f) data = f.getvalue() - logger.debug(f"caching {num_to_write} chunks (incremental={incremental}).") - chunks_to_write.clear() # free memory of the temporary table - if clear: - # if we don't need the in-memory chunks index anymore: - chunks.clear() # free memory, immediately - # the index object's name in the repo is the pure sha256 of its content, so borgstore can verify - # it the same way as any other object. an incompatible index format from a different borg version - # is rejected by borghash's own versioned header (MAGIC + VERSION) when it is read back. new_hash = hashlib.sha256(data).hexdigest() - if num_to_write == 0 and not force_write: - # don't persist an empty incremental index: if it became the only index/* (e.g. right - # after delete_chunkindex_from_repo()), build_chunkindex_from_repo() would return it as-is - # instead of rebuilding from the repo. with nothing new, the existing index is already up to date. - logger.debug("no new chunks to persist; not writing an empty incremental chunk index.") - return new_hash - stored_hashes = list_chunkindex_hashes(repository) + stored = False if force_write or new_hash not in stored_hashes: - # an index object is stored as index/, where is the sha256 of its content. - # when a client loads an index object, it compares the content hash against the hash in its - # name. if it is the same, the object is valid. if it is different, it is either corrupted or - # out of date and has to be discarded. when some functionality is DELETING chunks from the - # repository, it has to delete all existing index/* and maybe write a new, valid index/, - # so that all clients will discard any client-local chunks index caches. index_name = f"index/{new_hash}" logger.debug(f"storing chunks index as {index_name} in repository...") repository.store_store(index_name, data) + stored = True + return new_hash, stored + + +def write_chunkindex_to_repo( + repository, chunks, *, incremental=True, clear=False, force_write=False, delete_other=False, delete_these=None +): + # incremental controls *which* entries we write: only the F_NEW ones (a backup's new chunks) when + # True, else the whole index. borghash cannot serialize just the F_NEW entries, so either way we + # copy the selected entries into temporary table(s). + # Regardless of that, we always split the output into fragments of at most + # CHUNKINDEX_FRAGMENT_ENTRIES_MAX entries, so no single fragment gets too large (even the one + # incremental fragment of a huge initial backup). See repack_chunkindex for why we prefer many + # bounded, immutable fragments over one big index. + max_entries = CHUNKINDEX_FRAGMENT_ENTRIES_MAX + # the fragment set present in the repo before we start writing: + stored_hashes = set(list_chunkindex_hashes(repository)) + new_hashes = set() # content hashes of the fragments that make up the index we are writing now + stored_anything = False + fragments_written = 0 + + def flush(batch): + nonlocal stored_anything, fragments_written + count = len(batch) + if count == 0 and not force_write: + # don't persist an empty fragment: if it became the only index/* (e.g. right after + # delete_chunkindex_from_repo()), build_chunkindex_from_repo() would return it as-is + # instead of rebuilding from the repo. with nothing to write, the repo is already correct. + logger.debug("no new chunks to persist; not writing an empty chunk index fragment.") + batch.clear() + return + new_hash, stored = _store_chunkindex_fragment(repository, batch, stored_hashes, force_write=force_write) + batch.clear() # free memory of the temporary table + new_hashes.add(new_hash) + if stored: + stored_anything = True + fragments_written += 1 + + # sort the selected keys, so that an identical set of entries always produces identical + # fragments (identical content hashes), no matter in which order the entries were inserted + # into the hash table. this makes writing/repacking idempotent and convergent across clients: + # a fragment that already exists in the repo is not stored again (see _store_chunkindex_fragment) + # and no differently-partitioned duplicates of the same entries can pile up. + keys = sorted(key for key, _ in chunks.iteritems(only_new=incremental)) + total = len(keys) + # partition the selected entries into batches of at most max_entries entries and store each: + batch = ChunkIndex() + n = 0 + for key in keys: + # for now, we don't want to serialize the flags or the size: + batch[key] = chunks[key]._replace(flags=ChunkIndex.F_NONE, size=0) + n += 1 + if n >= max_entries: + flush(batch) + batch = ChunkIndex() + n = 0 + if n > 0 or total == 0: + # trailing, partially filled batch - or nothing to write at all: + # let flush handle force_write / the empty-index short-circuit. + flush(batch) + else: + batch.clear() # entries were an exact multiple of max_entries; no trailing batch to write + + logger.debug(f"cached {total} chunks (incremental={incremental}) in {fragments_written} fragment(s).") + if clear: + # if we don't need the in-memory chunks index anymore: + chunks.clear() # free memory, immediately + if stored_anything: # we have successfully stored to the repository, so we can clear all F_NEW flags now: chunks.clear_new() - # delete some no longer needed index objects, but never the one we just wrote: + + # delete some no longer needed index objects, but never the ones we just wrote. we only do this + # if we actually stored the replacement fragment(s), so we never leave the repo without an index. + if stored_anything and (delete_other or delete_these): if delete_other: - delete_these = set(stored_hashes) - {new_hash} - elif delete_these: - delete_these = set(delete_these) - {new_hash} + delete_these = set(stored_hashes) - new_hashes else: - delete_these = set() + delete_these = set(delete_these) - new_hashes for hash in delete_these: index_name = f"index/{hash}" try: @@ -599,7 +663,7 @@ def write_chunkindex_to_repo( pass if delete_these: logger.debug(f"chunk indexes deleted: {delete_these}") - return new_hash + return new_hashes def read_chunkindex_from_repo(repository, hash): @@ -619,6 +683,53 @@ def read_chunkindex_from_repo(repository, hash): logger.debug(f"{index_name} is invalid.") +def repack_chunkindex(repository): + """Consolidate small chunk-index fragments to keep their number and size in a healthy range. + + The chunks index lives in the repo as immutable, content-addressed index/ fragments. + Ordinary backups append a small incremental fragment each, so small fragments pile up over time. + This merges the small (< CHUNKINDEX_FRAGMENT_ENTRIES_MIN entries) fragments into fragments of up + to CHUNKINDEX_FRAGMENT_ENTRIES_MAX entries and deletes the small sources. Fragments already in + range are left untouched, so they stay immutable (and, once index/ is cache-backed, stay cached + for every client instead of being invalidated by an all-in-one consolidation). + + Merging is deferred: we only act when we can seal at least one full fragment (the small entries + sum to >= MIN) or when too many small fragments have piled up (more than + CHUNKINDEX_SMALL_FRAGMENT_CAP), so we don't rewrite a slowly growing fragment on every backup. + """ + small = [ + (name, approx) + for name, approx in list_chunkindex_fragments(repository) + if approx < CHUNKINDEX_FRAGMENT_ENTRIES_MIN + ] + if len(small) < 2: + return # nothing to gain from merging zero or one fragment + small_total = sum(approx for _, approx in small) + if small_total < CHUNKINDEX_FRAGMENT_ENTRIES_MIN and len(small) <= CHUNKINDEX_SMALL_FRAGMENT_CAP: + # can't seal a full fragment yet and not too many have piled up: defer. + return + logger.debug(f"repacking {len(small)} small chunk index fragments (~{small_total} entries)...") + merged = ChunkIndex() + merged_hashes = [] + for name, _ in small: + fragment = read_chunkindex_from_repo(repository, name) + if fragment is None: + # gone or invalid (e.g. deleted by another client); just don't merge it. + continue + for k, v in fragment.items(): + merged[k] = v + merged_hashes.append(name) + fragment.clear() + if not merged_hashes: + return + # write the merged entries split into bounded (<= MAX) fragments and delete the small sources. + # write_chunkindex_to_repo never deletes a hash it just wrote, so a fragment whose content is + # unchanged by the merge survives rather than being deleted and re-created. + write_chunkindex_to_repo( + repository, merged, incremental=False, clear=True, force_write=True, delete_these=merged_hashes + ) + + def build_chunkindex_from_repo( repository, *, disable_caches=False, cache_immediately=False, init_flags=ChunkIndex.F_USED ): @@ -638,10 +749,13 @@ def build_chunkindex_from_repo( chunks_to_merge.clear() if merged > 0: if merged > 1 and cache_immediately: - # immediately update the index, so we don't have to merge these again: - write_chunkindex_to_repo(repository, chunks, clear=False, force_write=True, delete_these=hashes) - else: - chunks.clear_new() + # consolidate small fragments on the repo so they don't pile up. this is a + # bounded repack: it does not collapse large, already-sealed fragments, so those + # stay immutable (and cache-stable for other clients) rather than being re-uploaded. + repack_chunkindex(repository) + # merging set F_NEW on every entry (see ChunkIndex.__setitem__); clear it, the repo + # already holds these entries in its fragments. + chunks.clear_new() return chunks # if we didn't get anything from the cache, compute the ChunkIndex the slow way: logger.debug("rebuilding the chunk index from the repo the slow way...") @@ -699,9 +813,9 @@ def chunks(self): # each backup writes a small incremental index/* fragment (only its new chunks), # which is cheap. collapsing them all into one big fragment on every run would re-upload # the whole index and, with delete_other, invalidate every other client's fragments -- - # a multi-GB churn per run on a shared repo. fragment count is reclaimed by `borg compact` - # (build_chunkindex_from_repo with cache_immediately). a size/threshold-based policy that - # bounds the fragment count without re-uploading large fragments can be added later. + # a multi-GB churn per run on a shared repo. instead, fragment count is bounded by + # repack_chunkindex (a size/threshold-based policy that merges only small fragments and + # leaves large, already-sealed ones untouched), run on close() and by `borg compact`. return self._chunks def seen_chunk(self, id, size=None): @@ -858,6 +972,16 @@ def close(self): now = datetime.now(UTC) self._maybe_write_chunks_cache(now, force=True, clear=True) self._chunks = None # nothing there (cleared!) + # this run just appended a (small) incremental fragment; consolidate accumulated small + # fragments so their count/size stays in a healthy range (bounded repack, see the function). + # repack works on the repo's index/* fragments, so it is independent of the in-memory index. + # the archive and its incremental fragment are already durably stored above, so repack is + # optional maintenance: if it fails (e.g. a transient store error), warn and carry on + # rather than failing an already-committed backup. the next run will repack again. + try: + repack_chunkindex(self.repository) + except Exception as exc: + logger.warning(f"consolidating the chunk index fragments failed (will retry next time): {exc}") # the index we just cleared in-place is the same object the repository holds; drop the # repository's reference too, so a later .chunks access rebuilds it from the repo instead # of seeing a valid-looking but empty index (and so is_chunk_index_loaded reports False). diff --git a/src/borg/constants.py b/src/borg/constants.py index ce58aaade7..cd82509a8c 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -95,6 +95,19 @@ # repo.list() result count limit used by the Borg client LIST_SCAN_LIMIT = 100000 +# The chunks index is stored in the repo as immutable, content-addressed index/ fragments. +# We keep each fragment's entry count within [MIN, MAX] where possible: MAX bounds a fragment's size, +# MIN keeps the fragment count down. Small (< MIN) fragments are merged (repacked); fragments already +# in range are left untouched (stable/immutable). SMALL_FRAGMENT_CAP bounds how many sub-MIN fragments +# may accumulate before we force a merge even if their entries still sum to less than MIN. +CHUNKINDEX_FRAGMENT_ENTRIES_MIN = 100000 # ~8MB +CHUNKINDEX_FRAGMENT_ENTRIES_MAX = 400000 # ~32MB +CHUNKINDEX_SMALL_FRAGMENT_CAP = 15 +# Approximate on-disk bytes per serialized entry: 32 (key) + 48 (value: flags 4 + size 4 + pack_id 32 +# + obj_offset 4 + obj_size 4). Only used to estimate a fragment's entry count from its byte size, +# so we can classify fragments without loading them. +CHUNKINDEX_ENTRY_SIZE = 80 + FD_MAX_AGE = 4 * 60 # 4 minutes # Some bounds on segment / segment_dir indexes diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index 626cc062b3..ef7533648d 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -6,7 +6,20 @@ from .hashindex_test import H from .crypto.key_test import TestKey from ..archive import Statistics -from ..cache import AdHocWithFilesCache, FileCacheEntry, delete_chunkindex_from_repo, read_chunkindex_from_repo +from .. import cache as cache_mod +from ..cache import ( + AdHocWithFilesCache, + ChunksMixin, + FileCacheEntry, + build_chunkindex_from_repo, + delete_chunkindex_from_repo, + list_chunkindex_fragments, + list_chunkindex_hashes, + read_chunkindex_from_repo, + repack_chunkindex, + write_chunkindex_to_repo, +) +from ..hashindex import ChunkIndex, ChunkIndexEntry from ..crypto.key import AESOCBKey from ..helpers import safe_ns from ..helpers.msgpack import int_to_timestamp @@ -121,9 +134,6 @@ def test_chunkindex_cache_not_consolidated_on_access(tmp_path): on every access would re-upload the whole index and, with delete_other, invalidate every other client's fragments. Fragment count is reclaimed by `borg compact`, not on every read here. """ - from ..cache import ChunksMixin, write_chunkindex_to_repo, list_chunkindex_hashes - from ..hashindex import ChunkIndex, ChunkIndexEntry - repository_location = os.fspath(tmp_path / "repository") with Repository(repository_location, exclusive=True, create=True) as repository: # seed extra fragments on top of the empty one written at repo creation @@ -142,3 +152,210 @@ def test_chunkindex_cache_not_consolidated_on_access(tmp_path): assert len(list_chunkindex_hashes(repository)) == before # ... and the in-memory index still resolves every seeded chunk assert H(1) in index and H(2) in index + + +def _ci_key(i): + """A distinct 32-byte chunk id for entry number i.""" + return i.to_bytes(32, "big") + + +def _make_chunkindex(keys): + ci = ChunkIndex() + for k in keys: + ci[k] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, k, 0, 4) + return ci + + +def _seed_fragment(repository, first, count): + """Write a fresh index fragment holding entries [first, first+count) and return its keys.""" + keys = [_ci_key(i) for i in range(first, first + count)] + write_chunkindex_to_repo(repository, _make_chunkindex(keys), incremental=False, force_write=True) + return keys + + +def test_write_chunkindex_splits_full_write(tmp_path, monkeypatch): + """A non-incremental (full) write splits the index into fragments of at most MAX entries.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) # start from a known-empty fragment set + keys = [_ci_key(i) for i in range(7000)] + write_chunkindex_to_repo( + repository, _make_chunkindex(keys), incremental=False, force_write=True, delete_other=True + ) + frags = list_chunkindex_fragments(repository) + # 7000 entries split by MAX=3000 -> 3 fragments (3000 + 3000 + 1000) + counts = sorted(len(read_chunkindex_from_repo(repository, name)) for name, _ in frags) + assert counts == [1000, 3000, 3000] + assert all(c <= 3000 for c in counts) + + +def test_repack_defers_when_below_min_and_few_fragments(tmp_path, monkeypatch): + """Deferred merging: small fragments summing to < MIN and not exceeding the cap are left alone.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 5) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + for j in range(3): # 3 fragments * 100 = 300 entries < MIN, count 3 <= cap 5 + _seed_fragment(repository, j * 100, 100) + before = {name for name, _ in list_chunkindex_fragments(repository)} + repack_chunkindex(repository) + after = {name for name, _ in list_chunkindex_fragments(repository)} + assert after == before # nothing merged + + +def test_repack_seals_when_smalls_reach_min(tmp_path, monkeypatch): + """When small fragments sum to >= MIN they are merged into a sealed fragment.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 100) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + all_keys = [] + for j in range(5): # 5 * 300 = 1500 >= MIN + all_keys += _seed_fragment(repository, j * 300, 300) + assert len(list_chunkindex_fragments(repository)) == 5 + repack_chunkindex(repository) + frags = list_chunkindex_fragments(repository) + assert len(frags) == 1 # 1500 entries, one fragment (< MAX) + merged = read_chunkindex_from_repo(repository, frags[0][0]) + assert len(merged) == 1500 + assert set(merged) == set(all_keys) + + +def test_repack_cap_forces_merge_below_min(tmp_path, monkeypatch): + """More than SMALL_FRAGMENT_CAP tiny fragments are merged even though they sum to < MIN.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 5) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + all_keys = [] + for j in range(6): # 6 * 50 = 300 < MIN, but count 6 > cap 5 + all_keys += _seed_fragment(repository, j * 50, 50) + assert len(list_chunkindex_fragments(repository)) == 6 + repack_chunkindex(repository) + frags = list_chunkindex_fragments(repository) + assert len(frags) == 1 # merged into a single sub-MIN remainder + merged = read_chunkindex_from_repo(repository, frags[0][0]) + assert len(merged) == 300 + assert set(merged) == set(all_keys) + + +def test_write_chunkindex_splits_incremental_write(tmp_path, monkeypatch): + """Even an incremental write (a single backup's new chunks) is split into <= MAX fragments.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 500) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + keys = [_ci_key(i) for i in range(1200)] # all F_NEW -> written by the incremental path + write_chunkindex_to_repo(repository, _make_chunkindex(keys), incremental=True) + counts = sorted( + len(read_chunkindex_from_repo(repository, name)) for name, _ in list_chunkindex_fragments(repository) + ) + assert counts == [200, 500, 500] # 1200 split by MAX=500 + + +def test_write_chunkindex_deterministic_fragments(tmp_path, monkeypatch): + """Identical entry sets always produce identical fragments, regardless of insertion order. + + The write path sorts the keys before partitioning them into batches, so the fragment set + (content hashes) only depends on the entries, not on the hash table's iteration order. + This makes writing/repacking idempotent and convergent across clients. + """ + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 500) + + key_ints = list(range(1200)) + hashes = [] + for reverse in (False, True): # build the same index with different insertion orders + repository_location = os.fspath(tmp_path / f"repository{reverse}") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + keys = [_ci_key(i) for i in (reversed(key_ints) if reverse else key_ints)] + write_chunkindex_to_repo(repository, _make_chunkindex(keys), incremental=False, force_write=True) + frags = list_chunkindex_fragments(repository) + hashes.append({name for name, _ in frags}) + # keys are big-endian ints, so sorted key order == numeric order: the batches must + # hold exactly the ranges [0..499], [500..999], [1000..1199]. + ranges = sorted(sorted(read_chunkindex_from_repo(repository, name)) for name, _ in frags) + assert ranges == [ + [_ci_key(i) for i in range(0, 500)], + [_ci_key(i) for i in range(500, 1000)], + [_ci_key(i) for i in range(1000, 1200)], + ] + assert hashes[0] == hashes[1] # identical fragment sets from differently-ordered inputs + + +def test_close_consolidates_fragments_across_sessions(tmp_path, monkeypatch): + """End-to-end: repeated create-like sessions leave a bounded, consolidated set of fragments.""" + monkeypatch.setenv("BORG_PASSPHRASE", "test") + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 200) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 500) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 1000) + + loc = os.fspath(tmp_path / "repository") + with Repository(loc, exclusive=True, create=True) as repository: + key = AESOCBKey.create(repository, TestKey.MockArgs()) + Manifest(key, repository).write() + + all_ids = [] + for s in range(5): # each session adds 100 new chunks (< MIN), so fragments must be consolidated + with Repository(loc, exclusive=True) as repository: + manifest = Manifest.load(repository, key=key, operations=Manifest.NO_OPERATION_CHECK) + cache = AdHocWithFilesCache(manifest) + try: + for i in range(s * 100, s * 100 + 100): + cid = H(i) + all_ids.append(cid) + cache.add_chunk(cid, {}, b"data-%d" % i, stats=Statistics()) + finally: + cache.close() + repository.flush() + + with Repository(loc, exclusive=True) as repository: + frags = list_chunkindex_fragments(repository) + # without repack there would be one incremental fragment per session (plus creation's empty); + # repack consolidates the small ones as they accumulate, so we end up with fewer. + assert len(frags) < 5 + assert any(approx >= 200 for _, approx in frags) # at least one sealed (>= MIN) fragment + assert all(approx <= 500 + 8 for _, approx in frags) # none exceeds MAX (+ header slack) + index = build_chunkindex_from_repo(repository) + for cid in all_ids: + assert cid in index + + +def test_repack_leaves_sealed_untouched_and_reconstructs(tmp_path, monkeypatch): + """Sealed (>= MIN) fragments survive a repack; build_chunkindex_from_repo reconstructs the index.""" + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 100) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + sealed_keys = _seed_fragment(repository, 0, 2000) # >= MIN -> sealed + sealed_hashes = {name for name, _ in list_chunkindex_fragments(repository)} + assert len(sealed_hashes) == 1 + small_keys = [] + for j in range(3): # 3 * 400 = 1200 >= MIN -> will be merged + small_keys += _seed_fragment(repository, 2000 + j * 400, 400) + + repack_chunkindex(repository) + + frags = {name for name, _ in list_chunkindex_fragments(repository)} + assert sealed_hashes <= frags # the sealed fragment was not rewritten or deleted + assert len(frags) == 2 # sealed one + the merged small ones + + index = build_chunkindex_from_repo(repository) + assert len(index) == 2000 + 1200 + assert set(index) == set(sealed_keys + small_keys) From 44475f85a25fb58259f374431e0042bcbed0a89b Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Fri, 3 Jul 2026 15:42:04 +0200 Subject: [PATCH 2/9] Make repack idempotent: gate index deletion on presence, not upload write_chunkindex_to_repo gated deletion of superseded index fragments on whether it actually uploaded the replacement (stored_anything). Combined with repack's force_write=True, every repack re-uploaded byte-identical merged fragments; and had repack simply dropped force_write, a crash between store and delete would leave the sources undeleted forever (the re-derived fragments dedupe-skip, stored_anything stays False, deletion never runs) -- small fragments piling up and being re-merged on every run. Gate deletion on new_hashes (the fragment set making up the index just written) instead. A dedupe-skipped fragment means its content is verifiably already in the repo, so deleting the superseded fragments is safe; an empty write (new_hashes empty) still skips deletion, so we never leave the repo without an index. repack now passes force_write=False, gaining idempotence with no redundant uploads. Add a regression test covering the crash/concurrent-repack recovery case. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 22 +++++++++++++------ src/borg/testsuite/cache_test.py | 37 ++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 6 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index 75d336e05d..0fe7b825a1 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -648,9 +648,16 @@ def flush(batch): # we have successfully stored to the repository, so we can clear all F_NEW flags now: chunks.clear_new() - # delete some no longer needed index objects, but never the ones we just wrote. we only do this - # if we actually stored the replacement fragment(s), so we never leave the repo without an index. - if stored_anything and (delete_other or delete_these): + # delete some no longer needed index objects, but never the ones we just wrote. we gate this on + # new_hashes (the fragments that make up the index we just wrote), not on whether we actually + # uploaded them: a fragment can be dedupe-skipped because its content already exists in the repo + # (see _store_chunkindex_fragment), and in that case the replacement is verifiably present, so + # deleting the superseded fragments is still safe. this also makes repack idempotent -- if a + # previous repack crashed after storing but before deleting, the next one re-derives the same + # fragments, dedupe-skips the uploads, and still deletes the small sources. when there is nothing + # to write (new_hashes empty, e.g. an empty incremental index), we skip deletion so we never leave + # the repo without an index. + if new_hashes and (delete_other or delete_these): if delete_other: delete_these = set(stored_hashes) - new_hashes else: @@ -723,10 +730,13 @@ def repack_chunkindex(repository): if not merged_hashes: return # write the merged entries split into bounded (<= MAX) fragments and delete the small sources. - # write_chunkindex_to_repo never deletes a hash it just wrote, so a fragment whose content is - # unchanged by the merge survives rather than being deleted and re-created. + # force_write=False so a merged fragment whose content already exists in the repo is dedupe-skipped + # rather than re-uploaded; write_chunkindex_to_repo still deletes the small sources because deletion + # is gated on the fragment set being present, not on a fresh upload (so repack is idempotent and + # avoids redundant uploads). write_chunkindex_to_repo also never deletes a hash it just wrote, so a + # fragment whose content is unchanged by the merge survives rather than being deleted and re-created. write_chunkindex_to_repo( - repository, merged, incremental=False, clear=True, force_write=True, delete_these=merged_hashes + repository, merged, incremental=False, clear=True, force_write=False, delete_these=merged_hashes ) diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index ef7533648d..f6c1b98dad 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -359,3 +359,40 @@ def test_repack_leaves_sealed_untouched_and_reconstructs(tmp_path, monkeypatch): index = build_chunkindex_from_repo(repository) assert len(index) == 2000 + 1200 assert set(index) == set(sealed_keys + small_keys) + + +def test_repack_idempotent_deletes_sources_when_merged_exists(tmp_path, monkeypatch): + """A repack whose merged fragment already exists (dedupe-skipped) still deletes the small sources. + + This is the crash / concurrent-repack recovery case: a prior repack may have stored the merged + fragment but not gotten to delete the small sources. The next repack re-derives the same merged + content, dedupe-skips the upload (force_write=False), and must still remove the now-superseded + sources -- otherwise they would pile up and be re-merged on every run. Deletion is therefore gated + on the merged fragment being present in the repo, not on a fresh upload having happened. + """ + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MAX", 3000) + monkeypatch.setattr(cache_mod, "CHUNKINDEX_SMALL_FRAGMENT_CAP", 100) + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + delete_chunkindex_from_repo(repository) + all_keys = [] + for j in range(5): # 5 * 300 = 1500 >= MIN -> merges into one sealed fragment + all_keys += _seed_fragment(repository, j * 300, 300) + repack_chunkindex(repository) + sealed = {name for name, _ in list_chunkindex_fragments(repository)} + assert len(sealed) == 1 # the merged, sealed fragment + + # simulate a repack that stored the merged fragment but crashed before deleting the sources: + # the sealed fragment is present, and the same small sources exist again alongside it. + for j in range(5): + _seed_fragment(repository, j * 300, 300) + assert len(list_chunkindex_fragments(repository)) == 6 # sealed + 5 re-seeded small + + repack_chunkindex(repository) # merged content already exists -> upload is dedupe-skipped + + frags = {name for name, _ in list_chunkindex_fragments(repository)} + assert frags == sealed # sources deleted; only the (unchanged) sealed fragment remains + merged = read_chunkindex_from_repo(repository, next(iter(frags))) + assert set(merged) == set(all_keys) From 7c4523baf7a576cbcf250aa6bdf88d1374f0a53a Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Fri, 3 Jul 2026 18:10:52 +0200 Subject: [PATCH 3/9] files cache: drop entries referencing missing chunks The files cache stores each file's chunk list "compressed" to chunk-index hash-table positions, valid only while the index is unmutated. When a backup runs out of space and aborts, pending chunks are rolled back out of the index; a files-cache entry memorized earlier still points at a now-stale position, so decompress_entry's self.chunks[id] raises KeyError. This crashed close() while it was cleaning up the already-failed backup, masking the real ENOSPC error with a "Key not found" traceback. Guard the two call sites that resolve cached entries against the index, both mirroring the existing compress_entry KeyError handling in _read_files_cache: - _write_files_cache drops the unresolvable entry (the file is re-chunked next backup) instead of raising. - file_known_and_unchanged treats the file as unknown so it gets re-chunked. Reproduced on a space-limited ramdisk (borg create -> ENOSPC): the crash is gone and the genuine "No space left on device" error surfaces instead; the repo remains consistent (check ok) and recoverable (compact + backup ok). Add a regression test that memorizes a file, deletes its chunk from the index, and asserts neither a cache lookup nor the files-cache save raises. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 22 +++++++++++++-- src/borg/testsuite/cache_test.py | 47 ++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index 0fe7b825a1..77e7f096bf 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -407,8 +407,18 @@ def _write_files_cache(self, files): entries = 0 age_discarded = 0 race_discarded = 0 + broken_discarded = 0 for path_hash, entry in files.items(): - entry = self.decompress_entry(entry) + try: + entry = self.decompress_entry(entry) + except KeyError: + # the entry references a chunk that is no longer in the chunks index (e.g. + # after an aborted / out-of-space backup rolled back pending chunks). such an + # entry is unusable; drop it rather than crash while saving the cache — the + # file will simply be re-chunked in a future backup. this mirrors the + # compress_entry KeyError handling in _read_files_cache. + broken_discarded += 1 + continue if entry.age == 0: # current entries if max(timestamp_to_int(entry.ctime), timestamp_to_int(entry.mtime)) < discard_after: # Only keep files seen in this backup that old enough not to suffer race conditions @@ -430,6 +440,8 @@ def _write_files_cache(self, files): entries += 1 integrity_data = fd.integrity_data files_cache_logger.debug(f"FILES-CACHE-KILL: removed {age_discarded} entries with age >= TTL [{ttl}]") + if broken_discarded: + files_cache_logger.debug(f"FILES-CACHE-KILL: removed {broken_discarded} entries referencing missing chunks") t_str = datetime.fromtimestamp(discard_after / 1e9, UTC).isoformat() files_cache_logger.debug(f"FILES-CACHE-KILL: removed {race_discarded} entries with ctime/mtime >= {t_str}") files_cache_logger.debug(f"FILES-CACHE-SAVE: finished, {entries} remaining entries saved.") @@ -463,7 +475,13 @@ def file_known_and_unchanged(self, hashed_path, path_hash, st): files_cache_logger.debug("UNKNOWN: no file metadata in cache for: %r", hashed_path) return False, None # we know the file! - entry = self.decompress_entry(entry) + try: + entry = self.decompress_entry(entry) + except KeyError: + # the cached entry references a chunk that is no longer in the chunks index (e.g. after + # an aborted / out-of-space backup); treat the file as unknown so it gets re-chunked. + files_cache_logger.debug("UNKNOWN: cached entry references a missing chunk: %r", hashed_path) + return False, None if "s" in cache_mode and entry.size != st.st_size: files_cache_logger.debug("KNOWN-CHANGED: file size has changed: %r", hashed_path) return True, None diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index f6c1b98dad..f2a4bcf8dd 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -396,3 +396,50 @@ def test_repack_idempotent_deletes_sources_when_merged_exists(tmp_path, monkeypa assert frags == sealed # sources deleted; only the (unchanged) sealed fragment remains merged = read_chunkindex_from_repo(repository, next(iter(frags))) assert set(merged) == set(all_keys) + + +def test_files_cache_save_tolerates_missing_chunk(tmp_path, monkeypatch): + """A files-cache entry whose chunk vanished from the index is dropped, not fatal. + + When a backup runs out of space and aborts, pending chunks get rolled back out of the chunks + index. A files-cache entry memorized earlier still references such a chunk by its (now stale) + hash-table index position, so decompress_entry's `self.chunks[id]` raises KeyError. That must not + crash a cache lookup or the files-cache save (which runs during close(), while cleaning up the + already-failed backup) -- the entry is simply discarded and the file re-chunked next time. + Regression test for the ENOSPC 'Key not found' crash in decompress_entry. + """ + monkeypatch.setenv("BORG_PASSPHRASE", "test") + loc = os.fspath(tmp_path / "repository") + with Repository(loc, exclusive=True, create=True) as repository: + key = AESOCBKey.create(repository, TestKey.MockArgs()) + Manifest(key, repository).write() + + with Repository(loc, exclusive=True) as repository: + manifest = Manifest.load(repository, key=key, operations=Manifest.NO_OPERATION_CHECK) + # size+inode+ctime cache mode -> files cache active; archive_name -> names the cache file + cache = AdHocWithFilesCache(manifest, cache_mode="cis", archive_name="test") + try: + cid = H(1) + cache.chunks[cid] = ChunkIndexEntry(ChunkIndex.F_NEW, 4, cid, 0, 4) + realfile = tmp_path / "f" + realfile.write_bytes(b"data") + st = os.stat(realfile) + path_hash = H(100) + cache.memorize_file(os.fspath(realfile), path_hash, st, [(cid, 4)]) + + # while the chunk is present, the file resolves as known/unchanged + known, chunks = cache.file_known_and_unchanged(os.fspath(realfile), path_hash, st) + assert known and chunks + + # simulate the out-of-space rollback: the referenced chunk is gone from the index + del cache.chunks[cid] + + # a lookup must now report the file as unknown (re-chunk it), not raise KeyError + known, chunks = cache.file_known_and_unchanged(os.fspath(realfile), path_hash, st) + assert known is False and chunks is None + + # saving the files cache must drop the broken entry rather than raise KeyError + cache._write_files_cache(cache._files) + finally: + cache.close() + repository.flush() From 1877953119873a3deff18cc3aeaa591ec817f3dd Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Sun, 5 Jul 2026 01:51:19 +0200 Subject: [PATCH 4/9] cache: make full index write explicit in build_chunkindex_from_repo --- src/borg/cache.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index 77e7f096bf..c73ac62807 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -815,7 +815,9 @@ def build_chunkindex_from_repo( logger.debug(f"queried {num_chunks} chunk IDs in {duration} s, ~{speed}/s") if cache_immediately: # immediately update the index, so we only rarely have to do it the slow way: - write_chunkindex_to_repo(repository, chunks, clear=False, force_write=True, delete_other=True) + write_chunkindex_to_repo( + repository, chunks, incremental=False, clear=False, force_write=True, delete_other=True + ) return chunks From 3e341f1d7675ee1e51f28fc74d1a8b94743ed34e Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Mon, 6 Jul 2026 23:37:29 +0200 Subject: [PATCH 5/9] cache: never build a partially merged chunk index, retry on vanished fragment A concurrent repack_chunkindex (another client closing its cache, under a shared lock) stores a merged replacement fragment and then deletes the small source fragments. A reader that listed index/* before the replacement existed and loaded the fragments after the deletion would silently skip the vanished fragment and return a partially merged chunk index: chunks that exist in the repo become invisible (spurious ObjectNotFound on read, lost deduplication, duplicate chunk copies written into new packs). Make the fragment merge all-or-nothing: if any listed fragment fails to load, discard the partial merge, re-list and retry - the fresh listing contains the replacement fragment. If no complete, consistent fragment set can be read within CHUNKINDEX_MERGE_ATTEMPTS, fall back to the slow (but correct) index rebuild from the pack headers instead of returning an incomplete index. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 40 +++++++++++++------- src/borg/constants.py | 3 ++ src/borg/testsuite/cache_test.py | 65 ++++++++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 13 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index c73ac62807..a5bd75b9b2 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -21,7 +21,7 @@ from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX -from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE +from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE, CHUNKINDEX_MERGE_ATTEMPTS from .hashindex import ChunkIndex, ChunkIndexEntry from .helpers import get_cache_dir from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list @@ -763,20 +763,31 @@ def build_chunkindex_from_repo( ): # first, try to build a fresh, mostly complete chunk index from centrally cached chunk indexes: if not disable_caches: - hashes = list_chunkindex_hashes(repository) - if hashes: # we have at least one cached chunk index! - merged = 0 - chunks = ChunkIndex() # we'll merge all we find into this + # a concurrent repack_chunkindex (another client, shared lock) deletes the small fragments it + # merged - only after storing the merged replacement, so no entries are ever lost, but a + # fragment we just listed can vanish before we load it. the index must be built from ALL + # fragments or not at all: a partially merged index would miss chunks that exist in the repo + # (spurious ObjectNotFound, lost deduplication). so on a failed load, re-list and retry - + # the fresh listing contains the replacement fragment. if we cannot get a complete, consistent + # set (e.g. a persistently unreadable fragment), fall through to the slow rebuild instead. + for _ in range(CHUNKINDEX_MERGE_ATTEMPTS): + hashes = list_chunkindex_hashes(repository) + if not hashes: # no cached chunk index available + break + chunks = ChunkIndex() # we'll merge all fragments into this + complete = True for hash in hashes: chunks_to_merge = read_chunkindex_from_repo(repository, hash) - if chunks_to_merge is not None: - logger.debug(f"cached chunk index {hash} gets merged...") - for k, v in chunks_to_merge.items(): - chunks[k] = v - merged += 1 - chunks_to_merge.clear() - if merged > 0: - if merged > 1 and cache_immediately: + if chunks_to_merge is None: + logger.debug(f"cached chunk index {hash} vanished or is invalid, restarting the merge...") + complete = False + break + logger.debug(f"cached chunk index {hash} gets merged...") + for k, v in chunks_to_merge.items(): + chunks[k] = v + chunks_to_merge.clear() + if complete: + if len(hashes) > 1 and cache_immediately: # consolidate small fragments on the repo so they don't pile up. this is a # bounded repack: it does not collapse large, already-sealed fragments, so those # stay immutable (and cache-stable for other clients) rather than being re-uploaded. @@ -785,6 +796,9 @@ def build_chunkindex_from_repo( # already holds these entries in its fragments. chunks.clear_new() return chunks + chunks.clear() # free the partial merge before retrying + else: + logger.warning("could not read a complete set of cached chunk index fragments, rebuilding the index.") # if we didn't get anything from the cache, compute the ChunkIndex the slow way: logger.debug("rebuilding the chunk index from the repo the slow way...") chunks = ChunkIndex() diff --git a/src/borg/constants.py b/src/borg/constants.py index cd82509a8c..8d4fa458c0 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -107,6 +107,9 @@ # + obj_offset 4 + obj_size 4). Only used to estimate a fragment's entry count from its byte size, # so we can classify fragments without loading them. CHUNKINDEX_ENTRY_SIZE = 80 +# How often to restart merging the fragments into a chunk index when a listed fragment vanishes +# mid-merge (a concurrent repack replaced it). After that, fall back to the slow rebuild from packs. +CHUNKINDEX_MERGE_ATTEMPTS = 3 FD_MAX_AGE = 4 * 60 # 4 minutes diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index f2a4bcf8dd..3bdc39ff9d 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -127,6 +127,71 @@ def test_read_chunkindex_from_repo_missing(tmp_path): assert result is None +def test_build_chunkindex_retries_on_vanished_fragment(tmp_path): + """build_chunkindex_from_repo must re-list and retry when a fragment vanishes mid-merge. + + A concurrent repack_chunkindex (another client closing its cache, shared lock) stores a merged + replacement fragment and then deletes the small source fragments. A reader that listed the + fragments before the replacement existed and loads them after the deletion must not silently + skip the vanished fragment (that would yield an incomplete index: spurious ObjectNotFound for + existing chunks, lost deduplication) - it must restart from a fresh listing. + """ + from borgstore.store import ObjectNotFound as StoreObjectNotFound + + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + # two fragments, as left behind by two incremental backups + for h in (H(1), H(2)): + ci = ChunkIndex() + ci[h] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, h, 0, 4) + write_chunkindex_to_repo(repository, ci, incremental=False, force_write=True) + fragment_names = list_chunkindex_hashes(repository) + original_store_load = repository.store_load + raced = False + + def racing_store_load(name, **kwargs): + nonlocal raced + if not raced and name.startswith("index/"): + raced = True + # simulate a concurrent repack: store the merged replacement fragment first, + # then delete the sources, then fail this load (the fragment has vanished). + merged = ChunkIndex() + for h in (H(1), H(2)): + merged[h] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, h, 0, 4) + write_chunkindex_to_repo(repository, merged, incremental=False, force_write=True) + for frag in fragment_names: + repository.store_delete(f"index/{frag}") + raise StoreObjectNotFound(name) + return original_store_load(name, **kwargs) + + repository.store_load = racing_store_load + try: + chunks = build_chunkindex_from_repo(repository) + finally: + repository.store_load = original_store_load + assert raced # the simulated repack fired + # the retry picked up the replacement fragment: the index is complete + assert H(1) in chunks + assert H(2) in chunks + + +def test_build_chunkindex_no_partial_merge(tmp_path): + """A persistently unreadable fragment must cause a rebuild from the packs, never a partial index.""" + repository_location = os.fspath(tmp_path / "repository") + with Repository(repository_location, exclusive=True, create=True) as repository: + # a valid fragment ... + ci = ChunkIndex() + ci[H(1)] = ChunkIndexEntry(ChunkIndex.F_NEW, 0, H(1), 0, 4) + write_chunkindex_to_repo(repository, ci, incremental=False, force_write=True) + # ... and a corrupt one (content does not match its name), which never loads successfully + repository.store_store(f"index/{'e' * 64}", b"garbage") + chunks = build_chunkindex_from_repo(repository) + # merging must have been abandoned in favor of the slow rebuild from the (empty) packs + # namespace; with the old skip-and-continue behavior, H(1) would be present here. + assert H(1) not in chunks + assert len(chunks) == 0 + + def test_chunkindex_cache_not_consolidated_on_access(tmp_path): """ChunksMixin.chunks binds the repository index without collapsing the cached fragments. From 49b3f62d8ca6a16100d0c932916f3a01674b5186 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 8 Jul 2026 10:58:59 +0200 Subject: [PATCH 6/9] cache: derive chunk index fragment entry size from the actual layout Replace the hardcoded CHUNKINDEX_ENTRY_SIZE = 80 constant with chunkindex_fragment_entry_size(), which derives the approximate on-disk bytes per serialized entry from the actual ChunkIndex layout (32 byte key + struct-packed ChunkIndexEntryFormat). This tracks any future change to the entry format automatically, so fragment classification cannot silently drift and misclassify sealed fragments as small (or vice versa). Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 27 ++++++++++++++++++++++----- src/borg/constants.py | 6 ++---- 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index a5bd75b9b2..fef17946af 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -4,6 +4,7 @@ import os import shutil import stat +import struct from collections import namedtuple from datetime import UTC, datetime, timedelta from pathlib import Path @@ -21,8 +22,8 @@ from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX -from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_ENTRY_SIZE, CHUNKINDEX_MERGE_ATTEMPTS -from .hashindex import ChunkIndex, ChunkIndexEntry +from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_MERGE_ATTEMPTS +from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat from .helpers import get_cache_dir from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list from .helpers import format_file_size, safe_encode @@ -545,18 +546,34 @@ def list_chunkindex_hashes(repository): return hashes +def chunkindex_fragment_entry_size(): + """Approximate on-disk bytes per serialized chunks-index entry. + + Derived from the actual ChunkIndex layout instead of a hardcoded constant, so it tracks any change + to the entry format automatically: a fragment stores each entry as its 32-byte key followed by the + struct-packed value (ChunkIndexEntryFormat, little-endian as borghash serializes it). Only used to + estimate a fragment's entry count from its stored byte size (see list_chunkindex_fragments), so we + can classify fragments without loading them; the small fixed header is ignored (negligible for the + fragment sizes we care about). + """ + key_size = 32 # a chunk id is a 256bit / 32 byte hash + value_size = struct.calcsize("<" + "".join(ChunkIndexEntryFormat)) + return key_size + value_size + + def list_chunkindex_fragments(repository): """Like list_chunkindex_hashes, but also return each fragment's approximate entry count. - The entry count is estimated from the stored object's byte size (CHUNKINDEX_ENTRY_SIZE bytes - per entry), so we can classify fragments (small vs. sealed) without loading them. The estimate + The entry count is estimated from the stored object's byte size (chunkindex_fragment_entry_size() + bytes per entry), so we can classify fragments (small vs. sealed) without loading them. The estimate ignores the small fixed header, which is negligible for the fragment sizes we care about. Returns a list of (name, approx_entries) tuples, sorted by name. """ + entry_size = chunkindex_fragment_entry_size() fragments = [] for info in repository.store_list("index"): info = ItemInfo(*info) # RPC does not give namedtuple - fragments.append((info.name, info.size // CHUNKINDEX_ENTRY_SIZE)) + fragments.append((info.name, info.size // entry_size)) fragments.sort() return fragments diff --git a/src/borg/constants.py b/src/borg/constants.py index 8d4fa458c0..336f626947 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -103,10 +103,8 @@ CHUNKINDEX_FRAGMENT_ENTRIES_MIN = 100000 # ~8MB CHUNKINDEX_FRAGMENT_ENTRIES_MAX = 400000 # ~32MB CHUNKINDEX_SMALL_FRAGMENT_CAP = 15 -# Approximate on-disk bytes per serialized entry: 32 (key) + 48 (value: flags 4 + size 4 + pack_id 32 -# + obj_offset 4 + obj_size 4). Only used to estimate a fragment's entry count from its byte size, -# so we can classify fragments without loading them. -CHUNKINDEX_ENTRY_SIZE = 80 +# (the approximate on-disk bytes per serialized entry used to classify fragments by their byte size +# is derived from the actual entry layout, see cache.chunkindex_fragment_entry_size()) # How often to restart merging the fragments into a chunk index when a listed fragment vanishes # mid-merge (a concurrent repack replaced it). After that, fall back to the slow rebuild from packs. CHUNKINDEX_MERGE_ATTEMPTS = 3 From 33954f582cb173e97b82a01923a9c4866ffeba16 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 8 Jul 2026 11:05:17 +0200 Subject: [PATCH 7/9] cache: fold list_chunkindex_hashes into list_chunkindex_fragments Both helpers walked the index/ namespace with the same store_list + ItemInfo unwrap + sort. Make list_chunkindex_fragments the single primitive and derive list_chunkindex_hashes from it, so the two views of the fragment set cannot drift apart if the listing ever changes. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 27 ++++++++++++--------------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index fef17946af..bd645f9732 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -535,17 +535,6 @@ def memorize_file(self, hashed_path, path_hash, st, chunks): ) -def list_chunkindex_hashes(repository): - hashes = [] - for info in repository.store_list("index"): - info = ItemInfo(*info) # RPC does not give namedtuple - # in the index/ namespace, each object's name is the sha256 hash of its content. - hashes.append(info.name) - hashes = sorted(hashes) - logger.debug(f"chunk indexes: {hashes}") - return hashes - - def chunkindex_fragment_entry_size(): """Approximate on-disk bytes per serialized chunks-index entry. @@ -562,11 +551,13 @@ def chunkindex_fragment_entry_size(): def list_chunkindex_fragments(repository): - """Like list_chunkindex_hashes, but also return each fragment's approximate entry count. + """List the index/ fragments, returning each fragment's (name, approximate entry count). - The entry count is estimated from the stored object's byte size (chunkindex_fragment_entry_size() - bytes per entry), so we can classify fragments (small vs. sealed) without loading them. The estimate - ignores the small fixed header, which is negligible for the fragment sizes we care about. + This is the single primitive that walks the index/ namespace; list_chunkindex_hashes is a thin + wrapper over it. In that namespace each object's name is the sha256 hash of its content. The entry + count is estimated from the stored object's byte size (chunkindex_fragment_entry_size() bytes per + entry), so we can classify fragments (small vs. sealed) without loading them. The estimate ignores + the small fixed header, which is negligible for the fragment sizes we care about. Returns a list of (name, approx_entries) tuples, sorted by name. """ entry_size = chunkindex_fragment_entry_size() @@ -578,6 +569,12 @@ def list_chunkindex_fragments(repository): return fragments +def list_chunkindex_hashes(repository): + hashes = [name for name, _ in list_chunkindex_fragments(repository)] # already sorted by name + logger.debug(f"chunk indexes: {hashes}") + return hashes + + def delete_chunkindex_from_repo(repository): hashes = list_chunkindex_hashes(repository) for hash in hashes: From f37bfa4343843b326dc97cc97e7750cc222cb074 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 8 Jul 2026 11:39:21 +0200 Subject: [PATCH 8/9] cache: partition chunk index write batches with chunkit Replace the hand-rolled batching in write_chunkindex_to_repo (manual n counter, flush closure with nonlocals, and the exact-multiple trailing special case) with helpers.chunkit, and drop the redundant stored_anything flag in favor of fragments_written. Pre-size each batch table via ChunkIndex(usable=...) so filling a full 400k-entry fragment does not repeatedly rehash. The empty force_write fragment stays an explicit case, as chunkit yields nothing for an empty input. Co-Authored-By: Claude Fable 5 --- src/borg/cache.py | 59 +++++++++++++++++++---------------------------- 1 file changed, 24 insertions(+), 35 deletions(-) diff --git a/src/borg/cache.py b/src/borg/cache.py index bd645f9732..6550975660 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -25,6 +25,7 @@ from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_MERGE_ATTEMPTS from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat from .helpers import get_cache_dir +from .helpers import chunkit from .helpers import hex_to_bin, bin_to_hex, parse_stringified_list from .helpers import format_file_size, safe_encode from .helpers import safe_ns @@ -627,26 +628,8 @@ def write_chunkindex_to_repo( # the fragment set present in the repo before we start writing: stored_hashes = set(list_chunkindex_hashes(repository)) new_hashes = set() # content hashes of the fragments that make up the index we are writing now - stored_anything = False fragments_written = 0 - def flush(batch): - nonlocal stored_anything, fragments_written - count = len(batch) - if count == 0 and not force_write: - # don't persist an empty fragment: if it became the only index/* (e.g. right after - # delete_chunkindex_from_repo()), build_chunkindex_from_repo() would return it as-is - # instead of rebuilding from the repo. with nothing to write, the repo is already correct. - logger.debug("no new chunks to persist; not writing an empty chunk index fragment.") - batch.clear() - return - new_hash, stored = _store_chunkindex_fragment(repository, batch, stored_hashes, force_write=force_write) - batch.clear() # free memory of the temporary table - new_hashes.add(new_hash) - if stored: - stored_anything = True - fragments_written += 1 - # sort the selected keys, so that an identical set of entries always produces identical # fragments (identical content hashes), no matter in which order the entries were inserted # into the hash table. this makes writing/repacking idempotent and convergent across clients: @@ -654,29 +637,35 @@ def flush(batch): # and no differently-partitioned duplicates of the same entries can pile up. keys = sorted(key for key, _ in chunks.iteritems(only_new=incremental)) total = len(keys) - # partition the selected entries into batches of at most max_entries entries and store each: - batch = ChunkIndex() - n = 0 - for key in keys: - # for now, we don't want to serialize the flags or the size: - batch[key] = chunks[key]._replace(flags=ChunkIndex.F_NONE, size=0) - n += 1 - if n >= max_entries: - flush(batch) - batch = ChunkIndex() - n = 0 - if n > 0 or total == 0: - # trailing, partially filled batch - or nothing to write at all: - # let flush handle force_write / the empty-index short-circuit. - flush(batch) + if keys: + # partition the selected entries into batches of at most max_entries entries: + batches = chunkit(keys, max_entries) + elif force_write: + # write a single empty fragment (e.g. at repo creation or after delete_chunkindex_from_repo()): + batches = [[]] else: - batch.clear() # entries were an exact multiple of max_entries; no trailing batch to write + # don't persist an empty fragment: if it became the only index/* (e.g. right after + # delete_chunkindex_from_repo()), build_chunkindex_from_repo() would return it as-is + # instead of rebuilding from the repo. with nothing to write, the repo is already correct. + logger.debug("no new chunks to persist; not writing an empty chunk index fragment.") + batches = [] + for batch_keys in batches: + # pre-size the temporary table to this batch so filling it does not repeatedly rehash: + batch = ChunkIndex(usable=len(batch_keys) or None) + for key in batch_keys: + # for now, we don't want to serialize the flags or the size: + batch[key] = chunks[key]._replace(flags=ChunkIndex.F_NONE, size=0) + new_hash, stored = _store_chunkindex_fragment(repository, batch, stored_hashes, force_write=force_write) + batch.clear() # free memory of the temporary table + new_hashes.add(new_hash) + if stored: + fragments_written += 1 logger.debug(f"cached {total} chunks (incremental={incremental}) in {fragments_written} fragment(s).") if clear: # if we don't need the in-memory chunks index anymore: chunks.clear() # free memory, immediately - if stored_anything: + if fragments_written: # we have successfully stored to the repository, so we can clear all F_NEW flags now: chunks.clear_new() From 4684af93fa1673d0ad2db8785c317e65f1e2ea4c Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 8 Jul 2026 11:45:23 +0200 Subject: [PATCH 9/9] hashindex: add ChunkIndexEntryFormat to the type stub chunkindex_fragment_entry_size() imports ChunkIndexEntryFormat from borg.hashindex; declare it (and its namedtuple type) in hashindex.pyi so mypy resolves the attribute. Co-Authored-By: Claude Fable 5 --- src/borg/hashindex.pyi | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/borg/hashindex.pyi b/src/borg/hashindex.pyi index 7dc26c3c8c..16795b7ccd 100644 --- a/src/borg/hashindex.pyi +++ b/src/borg/hashindex.pyi @@ -14,6 +14,15 @@ class ChunkIndexEntry(NamedTuple): CIE = tuple[int, int, bytes, int, int] | type[ChunkIndexEntry] +class ChunkIndexEntryFormatT(NamedTuple): + flags: str + size: str + pack_id: str + obj_offset: str + obj_size: str + +ChunkIndexEntryFormat: ChunkIndexEntryFormatT + class ChunkIndex: F_NONE: int F_USED: int