Read rsync output while it runs

Read the output as it arrives, split on newlines and carriage returns,
instead of all at once with communicate(). This is needed for the
progress bar. Invalid UTF-8 in the output is replaced instead of raising.

simple_backup/simple_backup.py

@@ -18,7 +18,9 @@ from __future__ import annotations
 import sys
 import os
 import pwd
-from typing import Callable, List, Optional, ParamSpec, TypedDict, TypeVar, Union
+import io
+import re
+from typing import Callable, Iterator, List, Optional, ParamSpec, TypedDict, TypeVar
 import warnings
 from functools import wraps
 from shutil import rmtree, which
@@ -647,24 +649,24 @@ class Backup:
             args.append('--rsync-path=sudo rsync')
 
         with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False) as p:
-            output: Union[bytes, List[str]]
+            assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase)
-            output, _ = p.communicate()
+            p.stdin.close()
+
+            output = [r for r in _read_records(p.stdout) if r != '']
 
             try:
                 del os.environ['SSHPASS']
             except KeyError:
                 pass
 
-            returncode = p.returncode
+            returncode = p.wait()
-
-        output = output.decode("utf-8").split('\n')
 
         if returncode == 0:
             if self._verbose:
                 logger.info('rsync: %s', output)
             else:
-                logger.info('rsync: %s', output[-3])
                 logger.info('rsync: %s', output[-2])
+                logger.info('rsync: %s', output[-1])
         else:
             self._returncode_log(returncode)
 
@@ -709,6 +711,21 @@ class Backup:
         return 0
 
 
+def _read_records(stream: io.BufferedIOBase) -> Iterator[str]:
+    """Yield rsync output as it arrives, split on '\\n' and '\\r' (which ends progress updates)"""
+
+    buffer = b''
+
+    while chunk := stream.read1(65536):
+        *records, buffer = re.split(rb'[\r\n]', buffer + chunk)
+
+        for r in records:
+            yield r.decode('utf-8', errors='replace')
+
+    if buffer != b'':
+        yield buffer.decode('utf-8', errors='replace')
+
+
 def _detect_user() -> Optional[str]:
     if os.geteuid() == 0:
         return os.getenv('SUDO_USER') or os.getenv('DOAS_USER')

tests/test_local.py

@@ -1,5 +1,6 @@
 """Tests for local backups, with real rsync runs and with a fake Popen"""
 
+import io
 import os
 from datetime import datetime, timedelta
 from pathlib import Path
@@ -195,6 +196,29 @@ class TestRsync:
         assert backups(out) == ['2026-10-14_11-00-00']
 
 
+class ChunkStream(io.BufferedIOBase):
+    """Returns the given chunks one per read1() call, like a pipe"""
+
+    def __init__(self, chunks: list[bytes]) -> None:
+        self.chunks = chunks
+
+    def read1(self, size: int = -1) -> bytes:
+        return self.chunks.pop(0) if self.chunks else b''
+
+
+class TestReadRecords:
+    def test_split_on_newline_and_cr(self) -> None:
+        stream = ChunkStream([b'file\n\r  1.00M  50%\r  2.00M 100%\n\nsent 1 bytes\ntotal'])
+
+        assert list(sb._read_records(stream)) == ['file', '', '  1.00M  50%', '  2.00M 100%', '', 'sent 1 bytes',
+                                                  'total']
+
+    def test_records_across_chunks(self) -> None:
+        stream = ChunkStream([b'fi', b'le\r', b'\nna', 'mé\n'.encode(), b'\xff\n'])
+
+        assert list(sb._read_records(stream)) == ['file', '', 'namé', '�']
+
+
 class FakePopen:
     """Stands in for rsync: creates the destination and returns a chosen return code"""
 
@@ -204,6 +228,8 @@ class FakePopen:
     def __init__(self, args: list[str], **kwargs: Any) -> None:
         FakePopen.calls.append(args)
         os.makedirs(args[args.index('/') + 1])
+        self.stdin = io.BytesIO()
+        self.stdout = io.BytesIO(b'sending incremental file list\nsent 1 bytes  received 2 bytes\ntotal size is 3\n')
 
     def __enter__(self) -> 'FakePopen':
         return self
@@ -211,8 +237,8 @@ class FakePopen:
     def __exit__(self, *args: Any) -> None:
         pass
 
-    def communicate(self) -> tuple[bytes, None]:
+    def wait(self) -> int:
-        return b'sending incremental file list\nsent 1 bytes  received 2 bytes\ntotal size is 3\n', None
+        return FakePopen.returncode
 
 
 @pytest.fixture

tests/test_remote.py

@@ -257,6 +257,8 @@ class RecordPopen:
 
     def __init__(self, args: list[str], **kwargs: Any) -> None:
         RecordPopen.calls.append(args)
+        self.stdin = io.BytesIO()
+        self.stdout = io.BytesIO(b'sent 1 bytes  received 2 bytes\ntotal size is 3\n')
 
     def __enter__(self) -> 'RecordPopen':
         return self
@@ -264,8 +266,8 @@ class RecordPopen:
     def __exit__(self, *args: Any) -> None:
         pass
 
-    def communicate(self) -> tuple[bytes, None]:
+    def wait(self) -> int:
-        return b'sent 1 bytes  received 2 bytes\ntotal size is 3\n', None
+        return RecordPopen.returncode
 
 
 @pytest.mark.usefixtures('connect')