From 8f01ad205a36b99613d64c817c55dafaac1b1d78 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Sun, 2 Aug 2026 16:43:45 +0200 Subject: [PATCH 1/2] 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 120a0f59ec..6c238db1da 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -299,11 +299,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] @@ -745,6 +753,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 34d39646fb..83ceb4f8b0 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -156,6 +156,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 c15f87ff711b6488b2ebfc08ebe05fedf701c502 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 12 Aug 2026 15:29:43 +0200 Subject: [PATCH 2/2] 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 638f772ce4..5f42185dca 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -152,6 +152,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 6c238db1da..d4ac1cc0af 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -266,10 +266,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 @@ -296,6 +300,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. @@ -305,16 +326,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. @@ -333,6 +347,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 @@ -753,13 +783,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): @@ -942,37 +975,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 83ceb4f8b0..a23d582c89 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -166,7 +166,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 @@ -175,34 +175,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): @@ -212,9 +253,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):