Repository: don't mask the original exception when unwinding with buffered chunks

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:
 

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: ...

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,17 +310,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.
@@ -335,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
@@ -891,7 +929,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):
@@ -1076,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)"""

src/borg/testsuite/repository_test.py

@@ -162,6 +162,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"