@@ -1513,6 +1513,19 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal,
                     # so it can be extracted / accessed in FUSE mount like a regular file.
                     # this needs to be done early, so that part files also get the patched mode.
                     item.mode = stat.S_IFREG | stat.S_IMODE(item.mode)
+
+                def rollback_chunks():
+                    # roll back the chunk references we added for the (still uncommitted) complete file item.
+                    # a read error can happen after we already added and referenced chunks (e.g. while reading
+                    # xattrs/flags or re-fstat-ing the file). if we then abort (and maybe retry) the file, those
+                    # references would leak (inflated refcount / orphan chunk). decref one reference per id in
+                    # item.chunks: for a cache hit this undoes the incref above, for a freshly chunked file it
+                    # undoes the add_chunk reference, and if part files were written it undoes the extra
+                    # "complete file" incref while leaving the (committed) part file references intact.
+                    for chunk in item.get('chunks', []):
+                        cache.chunk_decref(chunk.id, self.stats)
+                    item.chunks = []
+
                 if not hardlinked or hardlink_master:
                     if not is_special_file:
                         hashed_path = safe_encode(os.path.join(self.cwd, path))
@@ -1556,8 +1569,12 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal,
                         if is_win32:
                             changed_while_backup = False  # TODO
                         else:
-                            with backup_io('fstat2'):
-                                st2 = os.fstat(fd)
+                            try:
+                                with backup_io('fstat2'):
+                                    st2 = os.fstat(fd)
+                            except BackupOSError:
+                                rollback_chunks()
+                                raise
                             # special files:
                             # - fifos change naturally, because they are fed from the other side. no problem.
                             # - blk/chr devices don't change ctime anyway.
@@ -1581,9 +1598,7 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal,
                                 # not the last try and no part files written yet: trigger a retry by raising,
                                 # hoping that re-reading the file gives us a consistent copy. the retry is
                                 # done by the caller (Archiver._process_any).
-                                for chunk in item.chunks:
-                                    cache.chunk_decref(chunk.id, self.stats)
-                                item.chunks = []
+                                rollback_chunks()
                                 raise BackupError('file changed while we read it!')
                         if not is_special_file and not changed_while_backup:
                             # we must not memorize special files, because the contents of e.g. a
@@ -1592,7 +1607,13 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal,
                             # changed while we backed it up.
                             cache.memorize_file(hashed_path, path_hash, st, [c.id for c in item.chunks])
                     self.stats.nfiles += 1
-                item.update(self.metadata_collector.stat_ext_attrs(st, path, fd=fd))
+                try:
+                    item.update(self.metadata_collector.stat_ext_attrs(st, path, fd=fd))
+                except BackupOSError:
+                    # reading extended attributes (bsdflags / xattrs / ACLs) failed after we already
+                    # referenced the file's chunks - roll them back so a retry does not leak refcounts.
+                    rollback_chunks()
+                    raise
                 item.get_size(memorize=True)
                 return status
 
 
@@ -30,7 +30,7 @@
 import borg
 import borg.helpers.errors
 from .. import xattr, helpers, platform
-from ..archive import Archive, ChunkBuffer, ChunksProcessor
+from ..archive import Archive, ChunkBuffer, ChunksProcessor, MetadataCollector
 from ..archiver import Archiver, parse_storage_quota, PURE_PYTHON_MSGPACK_WARNING
 from ..cache import Cache, LocalCache
 from ..chunker import has_seek_hole
@@ -1325,6 +1325,45 @@ def test_create_erroneous_file_read_retry_rolls_back_chunks(self):
         for chunk_id, refs in item_chunk_refs.items():
             assert refcounts[chunk_id] == refs
 
+    def test_create_ext_attrs_error_retry_rolls_back_chunks(self):
+        # a read error that happens AFTER chunking (here: while reading the extended attributes) must
+        # also roll back the chunk increfs, so a retry does not end up with bad (too high) refcounts.
+        # the chunker reads the file fine; the failure is injected into stat_ext_attrs instead.
+        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)
+
+        orig_stat_ext_attrs = MetadataCollector.stat_ext_attrs
+        failed = []
+
+        def failing_stat_ext_attrs(self, st, path, fd=None):
+            # fail exactly once, on the first regular file we read the extended attributes for:
+            if stat.S_ISREG(st.st_mode) and not failed:
+                failed.append(True)
+                raise borg.helpers.errors.BackupOSError(
+                    'extended stat (xattrs)', OSError(errno.EIO, 'simulated I/O error', path))
+            return orig_stat_ext_attrs(self, st, path, fd=fd)
+
+        with patch.object(MetadataCollector, 'stat_ext_attrs', failing_stat_ext_attrs):
+            out = self.cmd('create', f'--chunker-params=fixed,{chunk_size}', '--list',
+                           self.repository_location + '::test', 'input')
+        assert 'retry: 1 of ' in out
+        assert failed  # the error was actually injected
+        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()}
+        # a leaked chunk from the failed 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
@@ -4560,6 +4599,10 @@ def test_create_changed_file_retry_rolls_back_chunks(self):
     def test_create_erroneous_file_with_part_files(self):
         pass
 
+    @unittest.skip('patches objects')
+    def test_create_ext_attrs_error_retry_rolls_back_chunks(self):
+        pass
+
     @unittest.skip('test_basic_functionality seems incompatible with fakeroot and/or the binary.')
     def test_basic_functionality(self):
         pass
@@ -5024,6 +5067,10 @@ def test_create_changed_file_retry_rolls_back_chunks(self):
     def test_create_erroneous_file_read_retry_rolls_back_chunks(self):
         pass
 
+    @unittest.skip('only works locally')  # inspects cache chunk refcounts via a local Cache/Repository
+    def test_create_ext_attrs_error_retry_rolls_back_chunks(self):
+        pass
+
     @unittest.skip('only works locally')
     def test_config(self):
         pass