Avoid blocking read on boundary CLI tool output

mycli/boundary_tunnel.py

@@ -44,6 +44,7 @@ def __init__(
         self.stdout = ''
         self._startup_error: OSError | ValueError | None = None
         self._started = threading.Event()
+        self._output_ready = threading.Event()
         self._ready = threading.Event()
         self._failed = threading.Event()
         self._thread: threading.Thread | None = None
@@ -71,19 +72,23 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
         self._thread.start()
         deadline = time.monotonic() + self.ready_timeout
         while time.monotonic() < deadline:
-            if self._failed.is_set():
-                self.close()
-                if self._startup_error is not None:
-                    raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
-                raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            self._raise_if_failed()
             if self._started.is_set():
-                self._read_stdout()
                 break
             time.sleep(0.05)
         else:
             self.close()
             raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process to start.')
 
+        while time.monotonic() < deadline:
+            self._raise_if_failed()
+            if self._output_ready.is_set():
+                break
+            time.sleep(0.05)
+        else:
+            self.close()
+            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process output.')
+
         connection_details = json.loads(self.stdout)
         self.username = connection_details['credentials'][0]['secret']['decoded']['username']
         self.password = connection_details['credentials'][0]['secret']['decoded']['password']
@@ -93,18 +98,22 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
         self.expiry = datetime.datetime.strftime(expiry_local, '%H:%M:%S %a %d %b %Y')
 
         while time.monotonic() < deadline:
-            if self._failed.is_set():
-                self.close()
-                if self._startup_error is not None:
-                    raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
-                raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            self._raise_if_failed()
             if self._is_listening():
                 self._ready.set()
                 return
             time.sleep(0.05)
         self.close()
         raise BoundaryTunnelError('Timed out waiting for Boundary tunnel to become ready.')
 
+    def _raise_if_failed(self) -> None:
+        if not self._failed.is_set():
+            return
+        self.close()
+        if self._startup_error is not None:
+            raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
+        raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+
     def _read_stdout(self) -> None:
         process = self.process
         if process is None or process.stdout is None:
@@ -141,6 +150,8 @@ def _run(self) -> None:
                 env=self._environment(),
             )
             self._started.set()
+            self._read_stdout()
+            self._output_ready.set()
         except (OSError, ValueError) as exc:
             self._startup_error = exc
             self._failed.set()

test/pytests/test_boundary_tunnel.py

@@ -3,6 +3,7 @@
 import os
 import socket
 import subprocess
+import threading
 from typing import Any, cast
 
 import pytest
@@ -100,8 +101,8 @@ def test_find_free_local_port_returns_available_port() -> None:
         sock.bind(('127.0.0.1', port))
 
 
-def test_boundary_tunnel_start_reads_stdout_in_main_thread(monkeypatch: pytest.MonkeyPatch) -> None:
-    calls: list[str] = []
+def test_boundary_tunnel_start_reads_stdout_in_worker_thread(monkeypatch: pytest.MonkeyPatch) -> None:
+    read_threads: list[str] = []
 
     class FakeProcess:
         returncode = 0
@@ -118,9 +119,11 @@ def wait(self, timeout: float | None = None) -> int:
     def fake_run() -> None:
         tunnel.process = cast(Any, FakeProcess())
         tunnel._started.set()
+        tunnel._read_stdout()
+        tunnel._output_ready.set()
 
     def fake_read_stdout() -> None:
-        calls.append('read_stdout')
+        read_threads.append(threading.current_thread().name)
         tunnel.stdout = CONNECTION_DETAILS
 
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406)
@@ -131,7 +134,7 @@ def fake_read_stdout() -> None:
 
     tunnel.start()
 
-    assert calls == ['read_stdout']
+    assert read_threads == ['mycli-boundary-tunnel']
     assert tunnel.stdout == CONNECTION_DETAILS
     assert tunnel.username == '1234'
     assert tunnel.password == '5678'
@@ -145,12 +148,10 @@ def test_boundary_tunnel_start_waits_for_process_to_start(monkeypatch: pytest.Mo
     def fake_sleep(seconds: float) -> None:
         sleeps.append(seconds)
         tunnel._started.set()
-
-    def fake_read_stdout() -> None:
         tunnel.stdout = CONNECTION_DETAILS
+        tunnel._output_ready.set()
 
     monkeypatch.setattr(tunnel, '_run', lambda: None)
-    monkeypatch.setattr(tunnel, '_read_stdout', fake_read_stdout)
     monkeypatch.setattr(tunnel, '_is_listening', lambda: True)
     monkeypatch.setattr(boundary_tunnel.time, 'sleep', fake_sleep)
 
@@ -188,6 +189,8 @@ def test_boundary_tunnel_start_reports_process_exit_after_stdout(monkeypatch: py
 
     def fake_run() -> None:
         tunnel._started.set()
+        tunnel._read_stdout()
+        tunnel._output_ready.set()
 
     def fake_read_stdout() -> None:
         tunnel.stdout = CONNECTION_DETAILS
@@ -205,6 +208,8 @@ def test_boundary_tunnel_start_reports_startup_error_after_stdout(monkeypatch: p
 
     def fake_run() -> None:
         tunnel._started.set()
+        tunnel._read_stdout()
+        tunnel._output_ready.set()
 
     def fake_read_stdout() -> None:
         tunnel.stdout = CONNECTION_DETAILS
@@ -251,16 +256,51 @@ def test_boundary_tunnel_start_reports_process_start_timeout(monkeypatch: pytest
         tunnel.start()
 
 
+def test_boundary_tunnel_start_reports_process_output_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[str] = []
+    output_released = threading.Event()
+
+    class BlockingStdout:
+        def readline(self) -> bytes:
+            calls.append('read')
+            output_released.wait()
+            return b''
+
+    class FakeProcess:
+        stdout = BlockingStdout()
+
+        def poll(self) -> None:
+            return None
+
+        def terminate(self) -> None:
+            calls.append('terminate')
+            output_released.set()
+
+        def wait(self, timeout: float | None = None) -> int:
+            calls.append(f'wait:{timeout}')
+            return 0
+
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'Popen', lambda *_args, **_kwargs: FakeProcess())
+    tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406, ready_timeout=0.1)
+
+    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel process output'):
+        tunnel.start()
+
+    assert calls[:2] == ['read', 'terminate']
+    assert sorted(calls[2:]) == ['wait:5', 'wait:None']
+
+
 def test_boundary_tunnel_start_reports_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406)
     tunnel.stdout = CONNECTION_DETAILS
 
     def fake_run() -> None:
         tunnel._started.set()
+        tunnel._output_ready.set()
 
     monkeypatch.setattr(tunnel, '_run', fake_run)
     monkeypatch.setattr(tunnel, '_is_listening', lambda: False)
-    monotonic_values = iter([0.0, 0.0, 31.0])
+    monotonic_values = iter([0.0, 0.0, 0.0, 31.0])
     monkeypatch.setattr(boundary_tunnel.time, 'monotonic', lambda: next(monotonic_values))
 
     with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel'):
@@ -310,6 +350,8 @@ def test_boundary_tunnel_run_tracks_process_status(
     popen_calls: list[tuple[list[str], dict[str, Any]]] = []
 
     class FakeProcess:
+        stdout = None
+
         def wait(self) -> int:
             return return_code