From d6c36e28af93927156468a356c479afc362423e7 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Sun, 2 Aug 2026 16:43:45 +0200 Subject: [PATCH 1/4] Repository: don't mask the original exception when unwinding with buffered chunks When a command aborts with chunks still buffered in the PackWriter, the "with repository:" unwind called close(), whose "PackWriter has unflushed chunks" assertion raised AssertionError and masked the original exception. Buffered chunks also left F_PENDING entries in the chunk index, which the close()-time index persist asserts on. This affects the paths that put chunks without a Cache: ArchiveChecker (borg check --repair) and borg debug put-obj. Commands that use a Cache are unaffected, because Cache.close() unwinds first and flushes the pack writer. ArchiveChecker.finish() flushes too, but only on the success path, so an abort before that still reaches close() with a non-empty buffer. Fix: on exception unwind, Repository.__exit__ drops the buffered pieces and their still-pending index entries via PackWriter._drop_buffered(), so the original exception propagates unmasked and no F_PENDING entries are persisted. The never-stored chunks die with the aborted operation. On a clean close, the assertion still catches a forgotten flush(). _drop_buffered() only ever runs while aborting, so it must not build the chunk index from the repo: that I/O can fail and mask the error being unwound. It now empties the buffer before it touches the index and skips the index cleanup when no index is loaded, where there is nothing to delete anyway. invalidate_chunk_index() is what leaves that state behind; its callers all flush first or never buffer, so this keeps the helper safe either way. Co-Authored-By: Claude Fable 5 --- src/borg/repository.py | 20 +++++++-- src/borg/testsuite/repository_test.py | 61 +++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/src/borg/repository.py b/src/borg/repository.py index 7b13fa11f8..a09c064f85 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -309,11 +309,19 @@ def _handoff(self): def _drop_buffered(self): """Drop the buffered pieces and their (still pending) index entries. - Called when a pack store failed: the caller is aborting, so chunks not yet handed - to the store die with it. Dropping their entries keeps the index free of F_PENDING - leftovers, like the sync store path does, so the close()-time index persist works. + Called when a pack store failed or the caller is unwinding an exception: the caller + is aborting, so chunks not yet handed to the store die with it. Dropping their + entries keeps the index free of F_PENDING leftovers, like the sync store path does, + so the close()-time index persist works. """ pieces = self._take_pieces() + if self.repository is not None and not self.repository.is_chunk_index_loaded: + # no in-memory index: the buffered chunks have no entries left to delete. going + # through self.chunks would build the index from the repo, and this helper only + # ever runs while aborting -- that I/O can fail and mask the error being unwound. + # invalidate_chunk_index() is what leaves this state behind; its callers all flush + # first or never buffer, so this keeps the helper safe either way. + return for chunk_id, _ in pieces: if chunk_id in self.chunks: # a chunk_id may appear more than once in the buffer del self.chunks[chunk_id] @@ -891,6 +899,12 @@ def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): + if exc_type is not None and self._pack_writer is not None: + # unwinding an exception: chunks still buffered in the pack writer were never + # stored, so they die with the aborted operation. drop them (and their + # F_PENDING index entries) so close() neither trips its flush assertion -- + # which would mask the original exception -- nor persists pending entries. + self._pack_writer._drop_buffered() self.close() @property diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 7fa3c4ab9f..4754455a51 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -162,6 +162,67 @@ def test_chunk_index_persisted_on_close(tmp_path): assert pdchunk(repository.get(H(x))) == b"DATA" +def test_exception_unwind_drops_buffered_chunks(tmp_path): + # An exception inside "with repository:" unwinds with chunks still buffered in the + # PackWriter (put() buffers until a pack fills or flush() is called). __exit__ must + # drop the buffered chunks so that close() neither replaces the original exception + # with its "call flush() before close()" assertion nor persists F_PENDING index + # entries for chunks that were never stored. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + assert repository._pack_writer._pieces # small chunk: still buffered, no pack written + raise ValueError("original error") + with Repository(location, exclusive=True) as repository: + # the buffered chunk died with the aborted operation: not in the index, not readable + assert H(0) not in repository.chunks + with pytest.raises(Repository.ObjectNotFound): + repository.get(H(0)) + + +def test_exception_unwind_does_not_rebuild_dropped_chunk_index(tmp_path, monkeypatch): + # Dropping the buffer runs only while aborting, so it must never build the chunk index + # from the repo: that I/O can fail and mask the error being unwound. With no in-memory + # index there is nothing to delete anyway. invalidate_chunk_index() is what leaves + # buffered chunks without an index; its callers all flush first or never buffer, so this + # test locks in the invariant rather than reproducing a reachable command path. + from .. import cache as cache_mod + + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + repository.flush() + + rebuilds = [] + + def must_not_rebuild(repository, *args, **kwargs): + rebuilds.append(1) + raise OSError("rebuilt the chunk index while unwinding") + + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True) as repository: + repository.put(H(1), fchunk(b"MORE")) + assert repository._pack_writer._pieces # still buffered, no pack written + repository.invalidate_chunk_index() # buffered chunks, no in-memory index + assert not repository.is_chunk_index_loaded + monkeypatch.setattr(cache_mod, "build_chunkindex_from_repo", must_not_rebuild) + raise ValueError("original error") + assert rebuilds == [] + + +def test_close_with_unflushed_chunks_asserts(tmp_path): + # On a clean (non-exception) path, closing with buffered chunks is a caller bug: + # the assertion in close() still catches a forgotten flush(). + location = os.fspath(tmp_path / "repo") + with pytest.raises(AssertionError, match="unflushed"): + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + # clean up the deliberately broken close: drop the buffered chunk, then close for real + repository._pack_writer._drop_buffered() + repository.close() + + def test_read_data(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: meta, data = b"meta", b"data" From 87d687d43c88095ecb7d9b365fc606bb7032aa86 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 12 Aug 2026 15:29:43 +0200 Subject: [PATCH 2/4] Repository: harden the abort path - join in-flight pack first, never raise from close() teardown Review follow-ups on the drop-on-unwind fix: Repository.__exit__ now calls PackWriter.discard(), the abort-side counterpart to flush(): it joins a still in-flight pack store first, so a pack that was already stored gets recorded in the index - and dropping buffered entries can no longer break update_pack_info() for a chunk id sitting in that pack and in the buffer (dropping first deleted the shared index entry, update_pack_info() then raised KeyError mid-pack and left F_PENDING leftovers for the close()-time persist to assert on, masking the original exception through a different door). Store errors from the join are logged, not raised. All abort-time index cleanup goes through PackWriter._drop_index_entries(): it never builds the chunk index from the repo (that I/O can fail and mask the error being unwound; with no in-memory index there is nothing to delete, since add() installs a chunk's entry before buffering its piece) and it only deletes entries that are still pending - a resolved entry means the chunk is in a stored pack, only the aborted duplicate piece dies. _apply_outcome() gets the same no-index guard, so joining a pack store while aborting cannot trigger a rebuild either. close() could still mask the original error a few lines further down: the close()-time chunk index persist and the lock release both do store I/O, which fails again exactly when the abort was caused by a failing store. Both are now logged instead of raised (the persisted index is only a cache; an unreleasable lock goes stale eventually), and the lock release and store close run in a finally block, so a close()-time error - e.g. the unflushed-chunks assertion - cannot leak the exclusive lock anymore. Also: log dropped buffer pieces (debug level), document the abort semantics in docs/internals/packs.rst, add ChunkIndex.is_pending/F_PENDING to the .pyi stub, new tests for the join-before-drop ordering and the guarded persist (both fail without the fixes). Co-Authored-By: Claude Fable 5 --- docs/internals/packs.rst | 4 + src/borg/hashindex.pyi | 2 + src/borg/repository.py | 148 +++++++++++++++++--------- src/borg/testsuite/repository_test.py | 66 +++++++++--- 4 files changed, 158 insertions(+), 62 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 21b5d33932..acbe7e6fac 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -198,6 +198,10 @@ The full ChunkIndex entry is ``(flags, size, pack_id, obj_offset, obj_size)`` (``ChunkIndexEntry`` in ``borg.hashindex``), where ``size`` is the plaintext chunk size. While a chunk is buffered in the pack writer but not yet flushed, its entry carries the ``F_PENDING`` flag and its pack location is unresolved. +When an operation aborts (an exception unwinds out of the repository context), +chunks still buffered in the pack writer were never stored: they are discarded +together with their pending index entries, while a pack already handed to the +store is still recorded if its store succeeded. .. _pack-write-order: diff --git a/src/borg/hashindex.pyi b/src/borg/hashindex.pyi index 8a9c43d8b3..313392ce1a 100644 --- a/src/borg/hashindex.pyi +++ b/src/borg/hashindex.pyi @@ -28,10 +28,12 @@ class ChunkIndex: F_USED: int F_COMPRESS: int F_NEW: int + F_PENDING: int M_USER: int M_SYSTEM: int def add(self, key: bytes, size: int) -> None: ... def update_pack_info(self, pack_results: list | None) -> None: ... + def is_pending(self, key: bytes) -> bool: ... def iteritems(self, *, only_new: bool = ..., prefix_bits: int = ..., prefix: int = ...) -> Iterator: ... @property def new_count(self) -> int: ... diff --git a/src/borg/repository.py b/src/borg/repository.py index a09c064f85..fe74b28782 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -276,10 +276,14 @@ def _apply_outcome(self, outcome): """ if outcome.error is not None: # the pack was not stored: drop the index entries for its chunks. - for chunk_id in outcome.pending_ids: - if chunk_id in self.chunks: # a chunk_id may appear more than once in this pack - del self.chunks[chunk_id] + self._drop_index_entries(outcome.pending_ids) raise outcome.error + if self.repository is not None and not self.repository.is_chunk_index_loaded: + # no in-memory index: this pack's entries died with it (see _drop_index_entries). + # do not build the index from the repo here: join_inflight also runs while closing + # or aborting, and that I/O could fail and mask an error being unwound. the stored + # pack is then simply not recorded, like the buffered pieces that die with an abort. + return outcome.results self.chunks.update_pack_info(outcome.results) # set the real location and clear F_PENDING return outcome.results @@ -306,6 +310,23 @@ def _handoff(self): self._inflight = (thread, outcome) thread.start() + def _drop_index_entries(self, chunk_ids): + """Drop the (still pending) index entries of *chunk_ids*, without building the index. + + Runs while aborting (a pack store failed, or the caller is unwinding an exception), + so it must never build the chunk index from the repo: that I/O can fail and mask the + error being unwound. No in-memory index means nothing to delete: add() installs a + chunk's index entry before buffering its piece, so pending entries never outlive a + dropped index. Entries that are not pending anymore are kept: their chunk is in a + stored pack, only the aborted (duplicate) piece dies. + """ + if self.repository is not None and not self.repository.is_chunk_index_loaded: + return + for chunk_id in chunk_ids: + # a chunk_id may appear more than once in a pack or buffer + if chunk_id in self.chunks and self.chunks.is_pending(chunk_id): + del self.chunks[chunk_id] + def _drop_buffered(self): """Drop the buffered pieces and their (still pending) index entries. @@ -315,16 +336,9 @@ def _drop_buffered(self): so the close()-time index persist works. """ pieces = self._take_pieces() - if self.repository is not None and not self.repository.is_chunk_index_loaded: - # no in-memory index: the buffered chunks have no entries left to delete. going - # through self.chunks would build the index from the repo, and this helper only - # ever runs while aborting -- that I/O can fail and mask the error being unwound. - # invalidate_chunk_index() is what leaves this state behind; its callers all flush - # first or never buffer, so this keeps the helper safe either way. - return - for chunk_id, _ in pieces: - if chunk_id in self.chunks: # a chunk_id may appear more than once in the buffer - del self.chunks[chunk_id] + if pieces: + logger.debug("dropping %d buffered chunk(s) while aborting", len(pieces)) + self._drop_index_entries(chunk_id for chunk_id, _ in pieces) def join_inflight(self): """Wait for an in-flight pack store and apply it to the index. @@ -343,6 +357,22 @@ def join_inflight(self): self._drop_buffered() raise + def discard(self): + """Join a still in-flight pack store, then drop the buffered pieces. + + The abort-side counterpart to flush(): a pack already handed to the store-thread is + joined first, so a stored pack gets recorded in the index and a failed one gets its + entries dropped; the pieces still buffered were never stored and die with the aborted + operation. Store errors are logged, not raised: the caller is aborting already, and + raising here would mask the error being unwound. + """ + try: + self.join_inflight() + except Exception as exc: + # join_inflight already dropped the failed pack's index entries and the buffer. + logger.warning("pack store failed while aborting: %s", exc) + self._drop_buffered() + def flush(self): """Write the current pack to the store. This is a barrier: any in-flight store is joined first and the current buffer is written synchronously, so afterwards @@ -899,13 +929,16 @@ def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): - if exc_type is not None and self._pack_writer is not None: - # unwinding an exception: chunks still buffered in the pack writer were never - # stored, so they die with the aborted operation. drop them (and their - # F_PENDING index entries) so close() neither trips its flush assertion -- - # which would mask the original exception -- nor persists pending entries. - self._pack_writer._drop_buffered() - self.close() + try: + if exc_type is not None and self._pack_writer is not None: + # unwinding an exception: chunks still buffered in the pack writer were never + # stored, so they die with the aborted operation. discard them (joining a + # still in-flight pack store first, so a stored pack gets recorded) so that + # close() neither trips its flush assertion -- which would mask the original + # exception -- nor persists pending index entries. + self._pack_writer.discard() + finally: + self.close() @property def id_str(self): @@ -1090,37 +1123,52 @@ def flush(self): self._pack_writer.flush() # PackWriter updates _chunks internally def close(self): - if self._pack_writer is not None: - try: - # normally a no-op: flush() is a barrier and runs before close(). when close() runs - # while unwinding an error, a pack store may still be in flight: join it, so a stored - # pack gets recorded in the index and a failed one gets its index entries dropped. - self._pack_writer.join_inflight() - except Exception as exc: - # do not raise: we are closing, probably unwinding an error already; raising here - # would just mask that original error. - logger.warning("pack store failed during close: %s", exc) - assert not self._pack_writer._pieces, "PackWriter has unflushed chunks; call flush() before close()" - # close() may run again after the store was already closed (idempotent close), so we can - # only persist while the store is open. Persisting is also a no-op unless chunks were added - # this session (only F_NEW entries are serialized, and an empty incremental write is skipped). - # guard on is_chunk_index_loaded so we never trigger a lazy rebuild just to persist on close. - if self.store_opened and self.is_chunk_index_loaded: - from .cache import write_chunkindex_to_repo + try: + if self._pack_writer is not None: + try: + # normally a no-op: flush() is a barrier and runs before close(). when close() runs + # while unwinding an error, a pack store may still be in flight: join it, so a stored + # pack gets recorded in the index and a failed one gets its index entries dropped. + self._pack_writer.join_inflight() + except Exception as exc: + # do not raise: we are closing, probably unwinding an error already; raising here + # would just mask that original error. + logger.warning("pack store failed during close: %s", exc) + assert not self._pack_writer._pieces, "PackWriter has unflushed chunks; call flush() before close()" + # close() may run again after the store was already closed (idempotent close), so we can + # only persist while the store is open. Persisting is also a no-op unless chunks were added + # this session (only F_NEW entries are serialized, and an empty incremental write is skipped). + # guard on is_chunk_index_loaded so we never trigger a lazy rebuild just to persist on close. + if self.store_opened and self.is_chunk_index_loaded: + from .cache import write_chunkindex_to_repo - write_chunkindex_to_repo(self, self.chunks, incremental=True) - if self.lock: - # ignore_not_found: close() runs during normal teardown, but also while unwinding an - # exception. if the lock was already gone (e.g. it went stale and another client killed - # it, or refresh() aborted with LockTimeout), a NotLocked raised here would mask the - # original error. we are closing anyway, so treat a missing lock as nothing to release. - self.lock.release(ignore_not_found=True) - self.lock = None - if self.store_opened: - self.store.close() - self.store_opened = False - self.opened = False - self._pack_cache.clear() + try: + write_chunkindex_to_repo(self, self.chunks, incremental=True) + except Exception as exc: + # do not raise: the persisted index is only a cache (rebuilt when missing or + # stale). close() often runs while unwinding a store error, and this persist + # writing to that same store would then raise again, masking the original error. + logger.warning("failed to persist the chunk index during close: %s", exc) + finally: + # release the lock and close the store even when the above raised (e.g. the unflushed- + # chunks assertion): a lock left behind would block other clients until it goes stale. + if self.lock: + # ignore_not_found: close() runs during normal teardown, but also while unwinding an + # exception. if the lock was already gone (e.g. it went stale and another client killed + # it, or refresh() aborted with LockTimeout), a NotLocked raised here would mask the + # original error. we are closing anyway, so treat a missing lock as nothing to release. + try: + self.lock.release(ignore_not_found=True) + except Exception as exc: + # do not raise: when the store is dead, the release fails, too -- raising would + # mask the original error, and the lock goes stale eventually anyway. + logger.warning("failed to release the lock during close: %s", exc) + self.lock = None + if self.store_opened: + self.store.close() + self.store_opened = False + self.opened = False + self._pack_cache.clear() def info(self): """return some infos about the repo (must be opened first)""" diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 4754455a51..a062bb634b 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -172,7 +172,7 @@ def test_exception_unwind_drops_buffered_chunks(tmp_path): with pytest.raises(ValueError, match="original error"): with Repository(location, exclusive=True, create=True) as repository: repository.put(H(0), fchunk(b"DATA")) - assert repository._pack_writer._pieces # small chunk: still buffered, no pack written + assert repository._pack_writer._pieces # still buffered: pack limits not reached raise ValueError("original error") with Repository(location, exclusive=True) as repository: # the buffered chunk died with the aborted operation: not in the index, not readable @@ -181,34 +181,75 @@ def test_exception_unwind_drops_buffered_chunks(tmp_path): repository.get(H(0)) +def test_exception_unwind_records_inflight_pack_drops_buffer(tmp_path): + # An exception unwinds while one pack is still in flight in the store-thread and more + # chunks sit in the buffer. __exit__ must join the in-flight store first -- recording + # the stored pack's chunks in the index -- and only drop what never reached a pack. + # H(0) is in the stored pack AND buffered again: its entry must survive, the chunk is + # stored; dropping it would first make update_pack_info() fail on the missing entry and + # then leave F_PENDING leftovers for the close()-time index persist to trip over. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + for x in range(3): # BORG_PACK_MAX_COUNT chunks (see conftest) fill a pack -> handed off + repository.put(H(x), fchunk(b"DATA")) + repository.put(H(0), fchunk(b"DATA")) # same id again: buffered + repository.put(H(3), fchunk(b"MORE")) # buffered + assert repository._pack_writer._pieces + raise ValueError("original error") + with Repository(location, exclusive=True) as repository: + for x in range(3): # the in-flight pack was stored: recorded in the index, readable + assert pdchunk(repository.get(H(x))) == b"DATA" + assert H(3) not in repository.chunks # the buffered chunk died with the abort + with pytest.raises(Repository.ObjectNotFound): + repository.get(H(3)) + + +def test_exception_unwind_survives_failing_index_persist(tmp_path, monkeypatch): + # close() persists the chunk index while unwinding an exception. when the abort was + # caused by the store failing, that persist fails, too -- it must be logged, not raised, + # so it cannot replace the original exception, and the lock still gets released. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + repository.flush() + + def broken_store(name, value): + raise OSError("store is dead") + + monkeypatch.setattr(repository.store, "store", broken_store) + raise ValueError("original error") + assert repository.lock is None # close() finished its teardown despite the failing persist + + def test_exception_unwind_does_not_rebuild_dropped_chunk_index(tmp_path, monkeypatch): # Dropping the buffer runs only while aborting, so it must never build the chunk index # from the repo: that I/O can fail and mask the error being unwound. With no in-memory - # index there is nothing to delete anyway. invalidate_chunk_index() is what leaves - # buffered chunks without an index; its callers all flush first or never buffer, so this - # test locks in the invariant rather than reproducing a reachable command path. + # index there is nothing to delete anyway: add() installs a chunk's index entry before + # buffering its piece, so pending entries never outlive a dropped index. from .. import cache as cache_mod location = os.fspath(tmp_path / "repo") - with Repository(location, exclusive=True, create=True) as repository: - repository.put(H(0), fchunk(b"DATA")) - repository.flush() + with Repository(location, exclusive=True, create=True): + pass rebuilds = [] def must_not_rebuild(repository, *args, **kwargs): rebuilds.append(1) - raise OSError("rebuilt the chunk index while unwinding") + return ChunkIndex() with pytest.raises(ValueError, match="original error"): with Repository(location, exclusive=True) as repository: repository.put(H(1), fchunk(b"MORE")) - assert repository._pack_writer._pieces # still buffered, no pack written + assert repository._pack_writer._pieces # still buffered: pack limits not reached repository.invalidate_chunk_index() # buffered chunks, no in-memory index assert not repository.is_chunk_index_loaded monkeypatch.setattr(cache_mod, "build_chunkindex_from_repo", must_not_rebuild) raise ValueError("original error") assert rebuilds == [] + assert not repository.is_chunk_index_loaded # the unwind never touched .chunks def test_close_with_unflushed_chunks_asserts(tmp_path): @@ -218,9 +259,10 @@ def test_close_with_unflushed_chunks_asserts(tmp_path): with pytest.raises(AssertionError, match="unflushed"): with Repository(location, exclusive=True, create=True) as repository: repository.put(H(0), fchunk(b"DATA")) - # clean up the deliberately broken close: drop the buffered chunk, then close for real - repository._pack_writer._drop_buffered() - repository.close() + # close()'s teardown runs in a finally block: even the failing close released the lock + # and closed the store, so nothing is left behind to clean up here. + assert repository.lock is None + assert not repository.store_opened def test_read_data(repo_fixtures, request): From e190f277ed8198c26041e6822dabedc87a465c8c Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Mon, 14 Sep 2026 18:03:22 +0200 Subject: [PATCH 3/4] ChunkIndex.add: keep a resolved pack location when re-adding a chunk Re-adding a chunk reset its entry to UNKNOWN/pending, dropping any prior location. For a re-put of an already stored chunk that meant: the chunk was unreadable until the next flush(), and when the operation aborted instead, the drop-on-unwind deleted the pending entry, so the close()-time persist saved a chunk index lacking a chunk that is in the store (review finding on this PR: put(x), flush(), put(x) again, abort -> x gone). Now add() keeps an existing resolved location: the chunk stays readable, an abort keeps its entry (the drop only deletes entries that are still pending), and a successful flush() overwrites the entry with the re-added copy's location via update_pack_info(), as before. Entries without a resolved location (new chunks, or re-adds while still pending) go to UNKNOWN/pending exactly as before. Co-Authored-By: Claude Fable 5 --- src/borg/hashindex.pyx | 19 +++++++++++++------ src/borg/testsuite/hashindex_test.py | 14 ++++++++++++++ 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/src/borg/hashindex.pyx b/src/borg/hashindex.pyx index a8764b724c..ba1757a328 100644 --- a/src/borg/hashindex.pyx +++ b/src/borg/hashindex.pyx @@ -97,12 +97,19 @@ class ChunkIndex(HTProxyMixin, MutableMapping): else: flags = v.flags | self.F_USED assert v.size == 0 or v.size == size - # F_PENDING marks the pack location (pack_id, obj_offset, obj_size) as not yet set. - # Re-adding a chunk resets it to UNKNOWN/pending, dropping any prior location until the next flush(). - self[key] = ChunkIndexEntry( - flags=flags | self.F_PENDING, size=size, - pack_id=UNKNOWN_BYTES32, obj_offset=UNKNOWN_INT32, obj_size=UNKNOWN_INT32 - ) + if v is not None and not (v.flags & self.F_PENDING): + # the chunk already has a resolved pack location: keep it, so the chunk stays readable + # and an abort can not lose a chunk that is already stored (#10013). the re-added + # copy's location replaces it at the next flush(), via update_pack_info(). + self[key] = ChunkIndexEntry( + flags=flags, size=size, pack_id=v.pack_id, obj_offset=v.obj_offset, obj_size=v.obj_size + ) + else: + # F_PENDING marks the pack location (pack_id, obj_offset, obj_size) as not yet set. + self[key] = ChunkIndexEntry( + flags=flags | self.F_PENDING, size=size, + pack_id=UNKNOWN_BYTES32, obj_offset=UNKNOWN_INT32, obj_size=UNKNOWN_INT32 + ) def __getitem__(self, key): """Specialized __getitem__ that hides system flags.""" diff --git a/src/borg/testsuite/hashindex_test.py b/src/borg/testsuite/hashindex_test.py index 7b56f93c33..387f5a8515 100644 --- a/src/borg/testsuite/hashindex_test.py +++ b/src/borg/testsuite/hashindex_test.py @@ -38,6 +38,20 @@ def test_chunkindex_add(): chunks.add(x, 3) # inconsistent size (we already have a different size) +def test_chunkindex_add_keeps_resolved_location(): + chunks = ChunkIndex() + x = H2(1) + chunks.add(x, 10) + pack_id = H2(2) + chunks.update_pack_info([(x, pack_id, 0, 50)]) + assert not chunks.is_pending(x) + # re-adding a stored chunk keeps its resolved location, so the chunk stays readable and an + # aborted re-put cannot lose it (#10013); the next flush() overwrites it via update_pack_info(). + chunks.add(x, 10) + assert not chunks.is_pending(x) + assert chunks[x] == ChunkIndexEntry(flags=ChunkIndex.F_USED, size=10, pack_id=pack_id, obj_offset=0, obj_size=50) + + def test_chunkindex_update_pack_info(): chunks = ChunkIndex() x1, x2 = H2(1), H2(2) From dedd8514401e88957f82373b4db954ca01383d84 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Mon, 14 Sep 2026 18:03:37 +0200 Subject: [PATCH 4/4] Repository.close: only swallow teardown errors while unwinding an exception close() wrapped the chunk index persist and the lock release in try/except unconditionally, so on a *clean* close a failing persist or lock release was only logged and the command still succeeded (review finding on this PR); on master both raise. close() takes an aborting=True keyword now, passed by __exit__ (and by __enter__'s failure path) when an exception is unwinding: only then are persist/release errors logged instead of raised, so they cannot mask the error being unwound. On a clean close they raise again, like on master -- but the lock release and store close still run in the finally block, so even a raising close does not leak the exclusive lock. A failing lock release is also only logged while the persist (or the unflushed- chunks assertion) is already raising, so teardown never masks an error. Also make the _apply_outcome() no-index skip unreachable on the normal path: flush() now asserts that the chunk index is loaded while chunks are buffered or in flight (invalidate_chunk_index() callers must flush first), so that caller bug fails loudly instead of silently leaving a stored pack unrecorded. Tests: cover the re-put-then-abort case, the failing in-flight store in discard(), a failing persist/lock release on both the clean and the unwinding path, an in-flight pack joined with a dropped index, and the new flush() assertion. The lock checks reopen the repository with an exclusive lock instead of asserting on self.lock, which close() used to reset either way. Co-Authored-By: Claude Fable 5 --- src/borg/repository.py | 56 ++++++++---- src/borg/testsuite/repository_test.py | 118 +++++++++++++++++++++++++- 2 files changed, 157 insertions(+), 17 deletions(-) diff --git a/src/borg/repository.py b/src/borg/repository.py index fe74b28782..dfad78a2fb 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -140,7 +140,9 @@ def build_rest_backend(location): class PackWriter: """Buffers chunks into a pack file and writes it to the store when full. - add() buffers a (chunk_id, cdata) pair and marks the chunk pending (F_PENDING); + add() buffers a (chunk_id, cdata) pair and marks the chunk pending (F_PENDING), unless + its index entry already has a resolved pack location (a re-put of a stored chunk keeps + that location, so the chunk stays readable, also if an abort drops the buffered copy); when the pack is full, it is built, hashed and stored, and each entry's pack_id, obj_offset and obj_size are set, clearing F_PENDING. @@ -283,6 +285,8 @@ def _apply_outcome(self, outcome): # do not build the index from the repo here: join_inflight also runs while closing # or aborting, and that I/O could fail and mask an error being unwound. the stored # pack is then simply not recorded, like the buffered pieces that die with an abort. + # flush() asserts the index is loaded while chunks are outstanding, so on the normal + # (non-aborting) path this state fails loudly there instead of being skipped here. return outcome.results self.chunks.update_pack_info(outcome.results) # set the real location and clear F_PENDING return outcome.results @@ -382,6 +386,13 @@ def flush(self): every chunk written by this flush (including a joined in-flight pack), or None if there was nothing to do. """ + # invalidating the chunk index with chunks buffered or in flight discards their entries, + # so this flush could not resolve their locations anymore: a caller must flush first. + assert ( + self.repository is None + or self.repository.is_chunk_index_loaded + or (self._inflight is None and not self._pieces) + ), "chunk index not loaded; flush() before invalidate_chunk_index()" results = self.join_inflight() or [] if self._pieces: pieces = self._take_pieces() @@ -924,7 +935,7 @@ def __enter__(self): try: self.open(exclusive=bool(self.exclusive), lock_wait=self.lock_wait, lock=self.do_lock) except Exception: - self.close() + self.close(aborting=True) raise return self @@ -938,7 +949,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): # exception -- nor persists pending index entries. self._pack_writer.discard() finally: - self.close() + self.close(aborting=exc_type is not None) @property def id_str(self): @@ -1122,7 +1133,14 @@ def flush(self): self._lock_refresh() self._pack_writer.flush() # PackWriter updates _chunks internally - def close(self): + def close(self, *, aborting=False): + """Close the repository: join an in-flight pack store, persist the chunk index, tear down. + + aborting=True means close() runs while unwinding an exception: teardown errors are then + logged instead of raised, so they cannot mask the error being unwound. On a clean close + (the default), a failing index persist or lock release raises, so the caller learns about + it -- but the lock release and store close still run, in the finally block. + """ try: if self._pack_writer is not None: try: @@ -1131,8 +1149,8 @@ def close(self): # pack gets recorded in the index and a failed one gets its index entries dropped. self._pack_writer.join_inflight() except Exception as exc: - # do not raise: we are closing, probably unwinding an error already; raising here - # would just mask that original error. + # do not raise: a store error here was already raised at the caller's put() or + # flush() if it cared; raising at close would mask an error being unwound. logger.warning("pack store failed during close: %s", exc) assert not self._pack_writer._pieces, "PackWriter has unflushed chunks; call flush() before close()" # close() may run again after the store was already closed (idempotent close), so we can @@ -1145,25 +1163,31 @@ def close(self): try: write_chunkindex_to_repo(self, self.chunks, incremental=True) except Exception as exc: - # do not raise: the persisted index is only a cache (rebuilt when missing or - # stale). close() often runs while unwinding a store error, and this persist - # writing to that same store would then raise again, masking the original error. + if not aborting: + raise + # unwinding, often a store error: this persist writing to that same store would + # raise again and mask it. the persisted index is only a cache (rebuilt when + # missing or stale), so losing this write costs a rebuild, not data. logger.warning("failed to persist the chunk index during close: %s", exc) finally: # release the lock and close the store even when the above raised (e.g. the unflushed- # chunks assertion): a lock left behind would block other clients until it goes stale. + # while any error is unwinding (aborting, or the try block above raised), a failing + # release is logged, not raised, so it cannot mask that error. + unwinding = aborting or sys.exc_info()[0] is not None if self.lock: - # ignore_not_found: close() runs during normal teardown, but also while unwinding an - # exception. if the lock was already gone (e.g. it went stale and another client killed - # it, or refresh() aborted with LockTimeout), a NotLocked raised here would mask the - # original error. we are closing anyway, so treat a missing lock as nothing to release. + # ignore_not_found: if the lock was already gone (e.g. it went stale and another + # client killed it, or refresh() aborted with LockTimeout), a NotLocked raised + # here would mask the original error; a missing lock is nothing to release. try: self.lock.release(ignore_not_found=True) + self.lock = None except Exception as exc: - # do not raise: when the store is dead, the release fails, too -- raising would - # mask the original error, and the lock goes stale eventually anyway. + if not unwinding: + raise + # when the store is dead, the release fails, too; the lock goes stale eventually. logger.warning("failed to release the lock during close: %s", exc) - self.lock = None + self.lock = None if self.store_opened: self.store.close() self.store_opened = False diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index a062bb634b..d36fe60ea6 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -205,6 +205,43 @@ def test_exception_unwind_records_inflight_pack_drops_buffer(tmp_path): repository.get(H(3)) +def test_reput_stored_chunk_survives_unwind(tmp_path): + # A re-put of an already stored chunk must not lose that chunk when the operation aborts: + # add() keeps the entry's resolved pack location, so the drop-on-unwind only deletes + # entries that never had one. Without that, the entry went back to pending, the drop + # deleted it, and the close()-time persist saved an index lacking a stored chunk. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + repository.flush() + repository.put(H(0), fchunk(b"DATA")) # same chunk again: buffered, entry stays resolved + raise ValueError("original error") + with Repository(location, exclusive=True) as repository: + assert pdchunk(repository.get(H(0))) == b"DATA" + + +def test_exception_unwind_survives_failing_inflight_store(tmp_path, monkeypatch): + # An exception unwinds while the in-flight pack store fails, too (e.g. the store died, + # which is why the operation aborted): discard() logs the store error instead of raising, + # so the original exception survives; the failed pack's index entries are dropped. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + + def broken_store(name, value): + raise OSError("store is dead") + + monkeypatch.setattr(repository.store, "store", broken_store) + for x in range(3): # fill a pack -> handed off; its store fails in the thread + repository.put(H(x), fchunk(b"DATA")) + repository.put(H(3), fchunk(b"MORE")) # still buffered + raise ValueError("original error") + with Repository(location, exclusive=True) as repository: + for x in range(4): # neither the failed pack's chunks nor the buffered one survived + assert H(x) not in repository.chunks + + def test_exception_unwind_survives_failing_index_persist(tmp_path, monkeypatch): # close() persists the chunk index while unwinding an exception. when the abort was # caused by the store failing, that persist fails, too -- it must be logged, not raised, @@ -220,7 +257,54 @@ def broken_store(name, value): monkeypatch.setattr(repository.store, "store", broken_store) raise ValueError("original error") - assert repository.lock is None # close() finished its teardown despite the failing persist + with Repository(location, exclusive=True): + pass # opens promptly with the exclusive lock: close() released it despite the failing persist + + +def test_clean_close_raises_on_failing_index_persist(tmp_path, monkeypatch): + # On a clean close (no exception unwinding), a failing index persist must raise, so the + # caller learns about it; the lock is still released by close()'s finally block. + location = os.fspath(tmp_path / "repo") + with pytest.raises(OSError, match="store is dead"): + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + repository.flush() + + def broken_store(name, value): + raise OSError("store is dead") + + monkeypatch.setattr(repository.store, "store", broken_store) + with Repository(location, exclusive=True): + pass # opens promptly with the exclusive lock: it was released despite the raise + + +def test_clean_close_raises_on_failing_lock_release(tmp_path, monkeypatch): + # On a clean close (no exception unwinding), a failing lock release must raise, so the + # caller learns the exclusive lock may still be in the repo. + location = os.fspath(tmp_path / "repo") + with pytest.raises(OSError, match="store is dead"): + with Repository(location, exclusive=True, create=True) as repository: + + def broken_release(*, ignore_not_found=False): + raise OSError("store is dead") + + monkeypatch.setattr(repository.lock, "release", broken_release) + repository.store.close() # the raise skipped the store close; tidy up + + +def test_exception_unwind_survives_failing_lock_release(tmp_path, monkeypatch): + # While unwinding an exception, a failing lock release is logged, not raised, so it + # cannot replace the original exception (when the store died, the release fails, too). + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + + def broken_release(*, ignore_not_found=False): + raise OSError("store is dead") + + monkeypatch.setattr(repository.lock, "release", broken_release) + raise ValueError("original error") + assert repository.lock is None # close() finished its teardown despite the failing release def test_exception_unwind_does_not_rebuild_dropped_chunk_index(tmp_path, monkeypatch): @@ -252,6 +336,38 @@ def must_not_rebuild(repository, *args, **kwargs): assert not repository.is_chunk_index_loaded # the unwind never touched .chunks +def test_exception_unwind_with_inflight_pack_and_dropped_index(tmp_path): + # An exception unwinds with a pack still in flight while the in-memory chunk index was + # dropped: joining must not rebuild the index from the repo (that I/O can fail and mask + # the error being unwound). The stored pack is simply not recorded -- its pending index + # entries died with the dropped index. + location = os.fspath(tmp_path / "repo") + with pytest.raises(ValueError, match="original error"): + with Repository(location, exclusive=True, create=True) as repository: + for x in range(3): # BORG_PACK_MAX_COUNT chunks (see conftest) fill a pack -> handed off + repository.put(H(x), fchunk(b"DATA")) + repository.invalidate_chunk_index() + assert not repository.is_chunk_index_loaded + raise ValueError("original error") + with Repository(location, exclusive=True) as repository: + # not recorded: the pack is in the store, but no index entry points into it + assert H(0) not in repository.chunks + + +def test_flush_after_invalidate_asserts(tmp_path): + # invalidate_chunk_index() with chunks buffered or in flight is a caller bug: their index + # entries died with the index, so a later flush() could not resolve their locations any + # more. flush() fails loudly instead of writing a pack no index entry points into. + location = os.fspath(tmp_path / "repo") + with Repository(location, exclusive=True, create=True) as repository: + repository.put(H(0), fchunk(b"DATA")) + repository.invalidate_chunk_index() + with pytest.raises(AssertionError, match="chunk index not loaded"): + repository.flush() + # discard the deliberately orphaned buffer, so the context can exit cleanly + repository._pack_writer.discard() + + def test_close_with_unflushed_chunks_asserts(tmp_path): # On a clean (non-exception) path, closing with buffered chunks is a caller bug: # the assertion in close() still catches a forgotten flush().