remote: deal with partial lines, fixes #2637

due to block buffering (in borg, pipes, sshd, ssh) partial lines might
be received. for plain text, this causes cosmetic issues, for json it
causes tracebacks due to parse errors.

the code now makes sure handle_remote_line() only gets called with a
complete line (which is terminated by any universal newline char, a
pure \r seems to be needed for remote progress displays).

it also fixes a yet undiscovered partial utf-8-sequence decoding issue
that might occur for the same reason.

src/borg/remote.py

@@ -520,6 +520,7 @@ class RemoteRepository:
         self.rx_bytes = 0
         self.tx_bytes = 0
         self.to_send = b''
+        self.stderr_received = b''  # incomplete stderr line bytes received (no \n yet)
         self.chunkid_to_msgids = {}
         self.ignore_responses = set()
         self.responses = {}
@@ -829,9 +830,16 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+.
                     if not data:
                         raise ConnectionClosed()
                     self.rx_bytes += len(data)
-                    data = data.decode('utf-8')
-                    for line in data.splitlines(keepends=True):
-                        handle_remote_line(line)
+                    # deal with incomplete lines (may appear due to block buffering)
+                    if self.stderr_received:
+                        data = self.stderr_received + data
+                        self.stderr_received = b''
+                    lines = data.splitlines(keepends=True)
+                    if lines and not lines[-1].endswith((b'\r', b'\n')):
+                        self.stderr_received = lines.pop()
+                    # now we have complete lines in <lines> and any partial line in self.stderr_received.
+                    for line in lines:
+                        handle_remote_line(line.decode('utf-8'))  # decode late, avoid partial utf-8 sequences
             if w:
                 while not self.to_send and (calls or self.preload_ids) and len(waiting_for) < MAX_INFLIGHT:
                     if calls:
@@ -962,6 +970,7 @@ def handle_remote_line(line):
 
     This function is remarkably complex because it handles multiple wire formats.
     """
+    assert line.endswith(('\r', '\n'))
     if line.startswith('{'):
         # This format is used by Borg since 1.1.0b6 for new-protocol clients.
         # It is the same format that is exposed by --log-json.

src/borg/testsuite/repository.py

@@ -944,16 +944,22 @@ class RemoteLoggerTestCase(BaseTestCase):
         sys.stderr = self.old_stderr
 
     def test_stderr_messages(self):
-        handle_remote_line("unstructured stderr message")
+        handle_remote_line("unstructured stderr message\n")
         self.assert_equal(self.stream.getvalue(), '')
         # stderr messages don't get an implicit newline
-        self.assert_equal(self.stderr.getvalue(), 'Remote: unstructured stderr message')
+        self.assert_equal(self.stderr.getvalue(), 'Remote: unstructured stderr message\n')
+
+    def test_stderr_progress_messages(self):
+        handle_remote_line("unstructured stderr progress message\r")
+        self.assert_equal(self.stream.getvalue(), '')
+        # stderr messages don't get an implicit newline
+        self.assert_equal(self.stderr.getvalue(), 'Remote: unstructured stderr progress message\r')
 
     def test_pre11_format_messages(self):
         self.handler.setLevel(logging.DEBUG)
         logging.getLogger().setLevel(logging.DEBUG)
 
-        handle_remote_line("$LOG INFO Remote: borg < 1.1 format message")
+        handle_remote_line("$LOG INFO Remote: borg < 1.1 format message\n")
         self.assert_equal(self.stream.getvalue(), 'Remote: borg < 1.1 format message\n')
         self.assert_equal(self.stderr.getvalue(), '')
 
@@ -961,7 +967,7 @@ class RemoteLoggerTestCase(BaseTestCase):
         self.handler.setLevel(logging.DEBUG)
         logging.getLogger().setLevel(logging.DEBUG)
 
-        handle_remote_line("$LOG INFO borg.repository Remote: borg >= 1.1 format message")
+        handle_remote_line("$LOG INFO borg.repository Remote: borg >= 1.1 format message\n")
         self.assert_equal(self.stream.getvalue(), 'Remote: borg >= 1.1 format message\n')
         self.assert_equal(self.stderr.getvalue(), '')
 
@@ -970,7 +976,7 @@ class RemoteLoggerTestCase(BaseTestCase):
         self.handler.setLevel(logging.WARNING)
         logging.getLogger().setLevel(logging.WARNING)
 
-        handle_remote_line("$LOG INFO borg.repository Remote: new format info message")
+        handle_remote_line("$LOG INFO borg.repository Remote: new format info message\n")
         self.assert_equal(self.stream.getvalue(), '')
         self.assert_equal(self.stderr.getvalue(), '')
 
@@ -990,7 +996,7 @@ class RemoteLoggerTestCase(BaseTestCase):
         foo_handler.setLevel(logging.INFO)
         logging.getLogger('borg.repository.foo').handlers[:] = [foo_handler]
 
-        handle_remote_line("$LOG INFO borg.repository Remote: new format child message")
+        handle_remote_line("$LOG INFO borg.repository Remote: new format child message\n")
         self.assert_equal(foo_stream.getvalue(), '')
         self.assert_equal(child_stream.getvalue(), 'Remote: new format child message\n')
         self.assert_equal(self.stream.getvalue(), '')