Unify chunk rollback in process_file_chunks

Demonstrator (per review request): move the "changed while we read it"
detection and reaction from FilesystemObjectProcessors.process_file down
into ChunksProcessor.process_file_chunks, so that both the read-error and
the changed-file rollback share a single rollback_uncommitted_chunks()
helper (decref item.chunks[from_chunk:], i.e. the tail not yet referenced
by a committed part item).

process_file_chunks now takes fd/st/is_special_file/files_changed/last_try
(keyword-only, default to "no change detection" for the pipe/tar/recreate
callers) and returns (part_number, changed_while_backup).

Tradeoff to evaluate: the rollback is now DRY and lives next to from_chunk,
but ChunksProcessor (a generic chunk-stream processor) gains fs-file stat
semantics (fstat2, ctime/mtime, files_changed) that arguably belong in
FilesystemObjectProcessors, and the method grows a tuple return.

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

Author: ThomasWaldmann

src/borg/archive.py

@@ -1309,7 +1309,8 @@ def maybe_checkpoint(self, item, from_chunk, part_number, forced=False):
                 logger.info('checkpoint requested: finished checkpoint creation!')
         return from_chunk, part_number
 
-    def process_file_chunks(self, item, cache, stats, show_progress, chunk_iter, chunk_processor=None):
+    def process_file_chunks(self, item, cache, stats, show_progress, chunk_iter, chunk_processor=None,
+                            *, fd=None, st=None, is_special_file=False, files_changed=None, last_try=False):
         if not chunk_processor:
             def chunk_processor(chunk):
                 chunk_id, data = cached_hash(chunk, self.key.id_hash)
@@ -1324,6 +1325,15 @@ def chunk_processor(chunk):
             del item.chunks_healthy
         from_chunk = 0
         part_number = 1
+
+        def rollback_uncommitted_chunks():
+            # roll back the chunks we added since the last checkpoint: they are not referenced by any
+            # (committed) part item, so they would otherwise leak (bad refcount / orphan chunk). the
+            # chunks before from_chunk are referenced by part files we already wrote, so we keep them.
+            for chunk in item.chunks[from_chunk:]:
+                cache.chunk_decref(chunk.id, stats)
+            item.chunks = item.chunks[:from_chunk]
+
         try:
             for chunk in chunk_iter:
                 item.chunks.append(chunk_processor(chunk))
@@ -1344,16 +1354,33 @@ def chunk_processor(chunk):
                         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]
+            # a read error happened after we already read+added some chunks: roll them back, then re-raise
+            # so the caller (Archiver._process_any) can retry or skip the file.
+            rollback_uncommitted_chunks()
             raise
-        # part_number > 1 means we wrote part files (checkpoints) for this file:
-        return part_number
+
+        # did the file change while we read it? this is only checked for real fs files, i.e. when the
+        # caller (FilesystemObjectProcessors.process_file) passed the file descriptor and stat result.
+        changed_while_backup = False
+        if fd is not None and not is_win32 and not is_special_file and files_changed != 'disabled':
+            # special files: fifos change naturally; blk/chr devices don't change ctime anyway.
+            with backup_io('fstat2'):
+                st2 = os.fstat(fd)
+            if files_changed == 'ctime':
+                changed_while_backup = st.st_ctime_ns != st2.st_ctime_ns
+            elif files_changed == 'mtime':
+                changed_while_backup = st.st_mtime_ns != st2.st_mtime_ns
+            else:
+                raise ValueError('invalid files_changed value: %r' % files_changed)
+        if changed_while_backup and not last_try and part_number == 1:
+            # regular file changed while we backed it up, might be inconsistent/corrupt! it is not the
+            # last try and we did not write part files yet, so we trigger a retry by raising (hoping to
+            # get a consistent copy on re-read). roll back the chunks we added, like the read-error path.
+            rollback_uncommitted_chunks()
+            raise BackupError('file changed while we read it!')
+        # changed_while_backup (with last_try or part files written) means: keep it, but flag as "C".
+        # part_number > 1 means we wrote part files (checkpoints) for this file.
+        return part_number, changed_while_backup
 
 
 class FilesystemObjectProcessors:
@@ -1543,42 +1570,19 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal,
                     if chunks is not None:
                         item.chunks = chunks
                     else:
+                        # process_file_chunks does the read+chunking, the checkpoint/part-file handling,
+                        # the "changed while we read it" detection and all the chunk rollback on errors
+                        # (read error or changed file). it raises to trigger a retry (see _process_any),
+                        # or returns changed_while_backup=True for the "keep but flag as C" case.
                         with backup_io('read'):
-                            part_number = self.process_file_chunks(
+                            part_number, changed_while_backup = self.process_file_chunks(
                                 item, cache, self.stats, self.show_progress,
-                                backup_io_iter(self.chunker.chunkify(None, fd)))
-                        if is_win32:
-                            changed_while_backup = False  # TODO
-                        else:
-                            with backup_io('fstat2'):
-                                st2 = os.fstat(fd)
-                            # special files:
-                            # - fifos change naturally, because they are fed from the other side. no problem.
-                            # - blk/chr devices don't change ctime anyway.
-                            if self.files_changed == 'disabled' or is_special_file:
-                                changed_while_backup = False
-                            elif self.files_changed == 'ctime':
-                                changed_while_backup = st.st_ctime_ns != st2.st_ctime_ns
-                            elif self.files_changed == 'mtime':
-                                changed_while_backup = st.st_mtime_ns != st2.st_mtime_ns
-                            else:
-                                raise ValueError('invalid files_changed value: %r' % self.files_changed)
+                                backup_io_iter(self.chunker.chunkify(None, fd)),
+                                fd=fd, st=st, is_special_file=is_special_file,
+                                files_changed=self.files_changed, last_try=last_try)
                         if changed_while_backup:
                             # regular file changed while we backed it up, might be inconsistent/corrupt!
-                            if last_try or part_number > 1:
-                                # this was the last try or we already wrote part files (checkpoints) for
-                                # this file. in both cases we must not retry (re-reading the file from the
-                                # start would create duplicate / inconsistent part files), so we keep what
-                                # we have and flag it as potentially inconsistent/corrupt.
-                                status = 'C'
-                            else:
-                                # 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 = []
-                                raise BackupError('file changed while we read it!')
+                            status = 'C'
                         if not is_special_file and not changed_while_backup:
                             # we must not memorize special files, because the contents of e.g. a
                             # block or char device will change without its mtime/size/inode changing.