From fa3263d77b007532448e0870fa63205757a37c52 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Fri, 3 Jul 2026 00:12:58 +0200 Subject: [PATCH] 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 4855baa57..75d336e05 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -20,6 +20,8 @@ files_cache_logger = create_logger("borg.debug.files_cache") 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 @@ class ChunksMixin: # 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 @@ class AdHocWithFilesCache(FilesCacheMixin, ChunksMixin): 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 ce58aaade..cd82509a8 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -95,6 +95,19 @@ MAX_ARCHIVES = 400000 # 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 626cc062b..ef7533648 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -6,7 +6,20 @@ import pytest 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)