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

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

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"