Merge pull request #3058 from ThomasWaldmann/fix-partial-json

remote: deal with partial lines, fixes #2637

src/borg/remote.py

@@ -521,6 +521,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 = {}
@@ -821,9 +822,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:
@@ -954,6 +962,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(), '')