Roll back chunks also on read errors (avoid bad refcounts)

The changed-file retry path already rolls back the chunks it added before
re-reading. Do the same when a read error (BackupOSError) interrupts reading
a file: the chunks added since the last checkpoint (item.chunks[from_chunk:])
are not referenced by any committed part item, so without rolling them back
they leak as an inflated chunk refcount (or, if content also shifted, an
orphan chunk).

The rollback lives in process_file_chunks, where from_chunk is known: chunks
before from_chunk are referenced by part files we already wrote and must be
kept; only the uncommitted tail is decref'd.

Adds test_create_erroneous_file_read_retry_rolls_back_chunks: the re-read
chunks dedup to the same ids, so the leak shows up as a too-high refcount
(not an orphan id) - the test checks refcounts directly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: ThomasWaldmann

src/borg/archive.py

@@ -1324,24 +1324,34 @@ def chunk_processor(chunk):
             del item.chunks_healthy
         from_chunk = 0
         part_number = 1
-        for chunk in chunk_iter:
-            item.chunks.append(chunk_processor(chunk))
-            if show_progress:
-                stats.show_progress(item=item, dt=0.2)
-            from_chunk, part_number = self.maybe_checkpoint(item, from_chunk, part_number, forced=False)
-        else:
-            if part_number > 1:
-                if item.chunks[from_chunk:]:
-                    # if we already have created a part item inside this file, we want to put the final
-                    # chunks (if any) into a part item also (so all parts can be concatenated to get
-                    # the complete file):
-                    from_chunk, part_number = self.maybe_checkpoint(item, from_chunk, part_number, forced=True)
-
-                # if we created part files, we have referenced all chunks from the part files,
-                # but we also will reference the same chunks also from the final, complete file:
-                for chunk in item.chunks:
-                    cache.chunk_incref(chunk.id, stats, size=chunk.size, part=True)
-                stats.nfiles_parts += part_number - 1
+        try:
+            for chunk in chunk_iter:
+                item.chunks.append(chunk_processor(chunk))
+                if show_progress:
+                    stats.show_progress(item=item, dt=0.2)
+                from_chunk, part_number = self.maybe_checkpoint(item, from_chunk, part_number, forced=False)
+            else:
+                if part_number > 1:
+                    if item.chunks[from_chunk:]:
+                        # if we already have created a part item inside this file, we want to put the final
+                        # chunks (if any) into a part item also (so all parts can be concatenated to get
+                        # the complete file):
+                        from_chunk, part_number = self.maybe_checkpoint(item, from_chunk, part_number, forced=True)
+
+                    # if we created part files, we have referenced all chunks from the part files,
+                    # but we also will reference the same chunks also from the final, complete file:
+                    for chunk in item.chunks:
+                        cache.chunk_incref(chunk.id, stats, size=chunk.size, part=True)
+                    stats.nfiles_parts += part_number - 1
+        except BackupOSError:
+            # a read error happened after we already read (and added to the repo/cache) some chunks.
+            # the chunks we added since the last checkpoint (item.chunks[from_chunk:]) are not referenced
+            # by any (committed) part item, so they would leak (bad refcount / orphan chunk) - roll them
+            # back. the chunks before from_chunk are referenced by part items we already wrote, keep them.
+            for chunk in item.chunks[from_chunk:]:
+                cache.chunk_decref(chunk.id, stats)
+            item.chunks = item.chunks[:from_chunk]
+            raise
         # part_number > 1 means we wrote part files (checkpoints) for this file:
         return part_number
 

src/borg/testsuite/archiver.py

@@ -1298,6 +1298,33 @@ def change_file_on_post_read_fstat(fd):
                 cached_ids = {id for id, entry in cache.chunks.iteritems()}
         assert cached_ids == referenced_ids
 
+    def test_create_erroneous_file_read_retry_rolls_back_chunks(self):
+        # a read error after we already read+added some chunks must roll those chunks back, so we do
+        # not end up with bad (too high) chunk refcounts. note: the re-read chunks dedup to the same
+        # ids, so a leak shows up as an inflated refcount (not as an orphan id), hence we check refcounts.
+        from collections import Counter
+        chunk_size = 1000
+        # distinct content per block, so each block maps to its own (unique) chunk id:
+        self.create_regular_file('file', contents=b'A' * chunk_size + b'B' * chunk_size)
+        self.cmd('init', '--encryption=repokey', self.repository_location)
+        # rErr: read 1st block ok, 2nd block fails -> retry; on the retry both blocks read ok.
+        out = self.cmd('create', f'--chunker-params=fail,{chunk_size},rErr', '--list',
+                       self.repository_location + '::test', 'input')
+        assert 'retry: 1 of ' in out
+        assert 'E input/file' not in out  # it was backed up ok on the retry
+
+        with Repository(self.repository_path, exclusive=True) as repository:
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
+            archive = Archive(repository, key, manifest, 'test')
+            item = [item for item in archive.iter_items() if item.path == 'input/file'][0]
+            item_chunk_refs = Counter(chunk.id for chunk in item.chunks)
+            with Cache(repository, key, manifest) as cache:
+                refcounts = {id: entry.refcount for id, entry in cache.chunks.iteritems()}
+        # each data chunk of the file must be referenced exactly as often as it occurs in the item -
+        # a leaked chunk from the failed read attempt would show up here as a too high refcount:
+        for chunk_id, refs in item_chunk_refs.items():
+            assert refcounts[chunk_id] == refs
+
     def test_create_erroneous_file_with_part_files(self):
         # if we have already written part files (checkpoints) for a file, a later read error must
         # NOT trigger a retry: re-reading the file from the start would create duplicate / inconsistent