upstream/borgbackup
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2026-07-15 21:12:50 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Repack chunks index into entry-count-bounded fragments

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Thomas Waldmann 2026-07-03 00:12:58 +02:00
parent d3a9799563
commit fa3263d77b
No known key found for this signature in database
GPG key ID: 243ACFA951F78E01
3 changed files with 407 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

222
src/borg/cache.py
View file

@ -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/<sha256> fragment.
We don't serialize the flags or the size, so callers pass entries with those zeroed. The object
is stored under index/<hash>, where <hash> 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/<hash>, where <hash> 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/<hash>,
# 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/<hash> 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).

13
src/borg/constants.py
View file

@ -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/<sha256> 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

225
src/borg/testsuite/cache_test.py
View file

@ -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)