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 120a0f59ec..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,17 +300,35 @@ 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. - 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() - 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. @@ -325,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 @@ -745,7 +783,16 @@ def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): - 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): @@ -928,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 34d39646fb..a23d582c89 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -156,6 +156,109 @@ 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 # 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 + assert H(0) not in repository.chunks + with pytest.raises(Repository.ObjectNotFound): + 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: 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): + pass + + rebuilds = [] + + def must_not_rebuild(repository, *args, **kwargs): + rebuilds.append(1) + 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: 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): + # 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")) + # 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): with get_repository_from_fixture(repo_fixtures, request) as repository: meta, data = b"meta", b"data"