remote: drain stderr in close(), avoid deadlock, fixes #10165

RemoteRepository.close() closed stdin and stdout and then did a bare
p.wait(). It abandoned the select loop that was draining the child's
stderr, but kept the read end of that pipe open - so if the child still
had more than a pipe buffer (64kiB) of stderr output to write, it
blocked writing to that pipe while borg blocked in wait4() waiting for
it to exit. Neither side could make progress and there was no timeout,
terminate() or kill() fallback, so the process pair hung forever.

close() now drains the remote's stderr until EOF - the remote's error
messages shall get to the client - and only then waits for the child,
with a timeout and SIGTERM / SIGKILL escalation, so that it can not
block unboundedly on a child that is blocked on a pipe we own.

(cherry picked from commit c204ad38d1a4b9fe6a08e5b93d38e6b5f2ec9ee5)

src/borg/remote.py

@@ -13,7 +13,7 @@ import tempfile
 import textwrap
 import time
 import traceback
-from subprocess import Popen, PIPE
+from subprocess import Popen, PIPE, TimeoutExpired
 
 from . import __version__
 from .compress import Compressor
@@ -44,6 +44,12 @@ MAX_INFLIGHT = 100
 
 RATELIMIT_PERIOD = 0.1
 
+# when shutting down the connection, we still want to receive (and log) the remote's stderr output,
+# but we must not wait for it forever if the remote is stuck or gone:
+STDERR_DRAIN_IDLE_TIMEOUT = 30.0  # seconds without any stderr data before we stop draining it
+# after that, wait for the child process to exit - and make it exit if it does not do so voluntarily:
+CHILD_EXIT_TIMEOUT = 10.0  # seconds to wait for the child, before SIGTERM and (again) before SIGKILL
+
 
 def os_write(fd, data):
     """os.write wrapper so we do not lose data for partial writes."""
@@ -857,17 +863,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+.
                     data = os.read(fd, 32768)
                     if not data:
                         raise ConnectionClosed()
-                    self.rx_bytes += len(data)
-                    # 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())  # decode late, avoid partial utf-8 sequences
+                    self.handle_stderr_data(data)
             if w:
                 while (len(self.to_send) <= maximum_to_send) and (calls or self.preload_ids) and len(waiting_for) < MAX_INFLIGHT:
                     if calls:
@@ -973,11 +969,56 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+.
     def break_lock(self):
         """actual remoting is done via self.call in the @api decorator"""
 
+    def handle_stderr_data(self, data):
+        """Log the remote's stderr output (given as bytes, maybe not ending on a line boundary)."""
+        self.rx_bytes += len(data)
+        # 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())  # decode late, avoid partial utf-8 sequences
+
+    def drain_stderr(self):
+        """Read and log the remote's stderr until EOF (or until it is idle for too long).
+
+        We are the only reader of that pipe, so we must keep reading it: if we did not, the child
+        could block forever writing into a full pipe buffer while we wait for it to terminate.
+        Also, the remote's stderr transports its error messages, so we want to see them.
+        """
+        while True:
+            r, w, x = select.select([self.stderr_fd], [], [], STDERR_DRAIN_IDLE_TIMEOUT)
+            if not r:
+                logger.debug('drain_stderr: remote stderr idle for %.1fs, giving up on it.',
+                             STDERR_DRAIN_IDLE_TIMEOUT)
+                break
+            data = os.read(self.stderr_fd, 32768)
+            if not data:
+                break  # EOF
+            self.handle_stderr_data(data)
+        if self.stderr_received:  # log a trailing partial line, if any
+            self.handle_stderr_data(b'\n')
+
     def close(self):
         if self.p:
             self.p.stdin.close()
             self.p.stdout.close()
-            self.p.wait()
+            self.drain_stderr()
+            self.p.stderr.close()
+            # the child should terminate now, but just in case it does not, do not wait forever:
+            for make_it_terminate in (self.p.terminate, self.p.kill):
+                try:
+                    self.p.wait(timeout=CHILD_EXIT_TIMEOUT)
+                    break
+                except TimeoutExpired:
+                    logger.warning('Remote process %d did not terminate, sending it a signal.', self.p.pid)
+                    make_it_terminate()
+            else:
+                self.p.wait()  # SIGKILLed, so this will not block for long
             self.p = None
 
     def async_response(self, wait=True):

src/borg/testsuite/remote.py

@@ -1,12 +1,15 @@
 import errno
 import os
 import io
+import sys
+import threading
 import time
+from subprocess import Popen, PIPE
 from unittest.mock import patch
 
 import pytest
 
-from ..remote import SleepingBandwidthLimiter, RepositoryCache, cache_if_remote
+from ..remote import RemoteRepository, SleepingBandwidthLimiter, RepositoryCache, cache_if_remote
 from ..repository import Repository
 from ..crypto.key import PlaintextKey
 from ..compress import CompressionSpec
@@ -199,3 +202,34 @@ class TestRepositoryCache:
 
         with pytest.raises(IntegrityError):
             assert next(iterator) == (7, b'5678')
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason='remote repositories are not supported on Windows')
+def test_close_drains_stderr():
+    """RemoteRepository.close() must not deadlock on a child that still writes to stderr, see #10165.
+
+    borg is the only reader of the child's stderr pipe, so if it stopped reading it while waiting
+    for the child to terminate, the child would block forever writing into the full pipe buffer.
+    """
+    # the child waits for EOF on stdin (like the remote does when we close our end),
+    # then writes way more than a pipe buffer (64kiB) worth of stderr output:
+    code = 'import sys; sys.stdin.read(); sys.stderr.write(("E" * 4095 + "\\n") * 64)'
+    p = Popen([sys.executable, '-c', code], bufsize=0, stdin=PIPE, stdout=PIPE, stderr=PIPE)
+    # set up just enough of a RemoteRepository for close() to work on:
+    rr = RemoteRepository.__new__(RemoteRepository)
+    rr.p = p
+    rr.stderr_fd = p.stderr.fileno()
+    rr.stderr_received = b''
+    rr.rx_bytes = 0
+    rr.responses = {}  # only used by RemoteRepository.__del__
+    os.set_blocking(rr.stderr_fd, False)
+    # if close() deadlocks, it never returns - so do not call it in the main thread:
+    closer = threading.Thread(target=rr.close, daemon=True)
+    closer.start()
+    closer.join(timeout=30)
+    if closer.is_alive():
+        p.kill()  # unblock the deadlocked close(), so the child does not stay around
+        pytest.fail('RemoteRepository.close() did not return, see #10165.')
+    assert rr.p is None
+    assert p.returncode == 0  # the child could write all its output and terminated by itself
+    assert rr.rx_bytes == 64 * 4096  # we have received all of the child's stderr output