Stop cancelled remote queries without blocking the selector

RemoteQueryPool.cancel() terminated the SSH process and then waited up
to 250 ms for it, on the curses thread. Ctrl+R re-enqueues every host,
so a refresh of six hosts could freeze the popup for over a second.

Only send SIGTERM when cancelling; reap the process on later polls,
escalating to SIGKILL after 250 ms and closing its pipes once it has
exited. close() still waits, since the popup is exiting anyway.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

herdr_hub/remotes.py

@@ -596,6 +596,13 @@ class _RemoteJob:
     posix_fallback: bool = False
 
 
+@dataclass
+class _StoppingProcess:
+    process: subprocess.Popen[bytes]
+    since: float
+    killed: bool = False
+
+
 class RemoteQueryPool:
     """Pollable bounded pool of SSH processes for an interactive selector."""
 
@@ -614,6 +621,7 @@ class RemoteQueryPool:
         self._pending: deque[RemoteEntry] = deque()
         self._active: dict[str, _RemoteJob] = {}
         self._completed: deque[RemoteQueryResult] = deque()
+        self._stopping: list[_StoppingProcess] = []
 
     def enqueue(self, entry: RemoteEntry) -> None:
         self.cancel(entry.target)
@@ -639,6 +647,7 @@ class RemoteQueryPool:
         )
 
     def poll(self) -> list[RemoteQueryResult]:
+        self._reap_stopping()
         finished = list(self._completed)
         self._completed.clear()
         now = time.monotonic()
@@ -689,6 +698,15 @@ class RemoteQueryPool:
         for job in list(self._active.values()):
             self._stop_process(job.process)
         self._active.clear()
+        # The popup is exiting: wait briefly so no zombie or open pipe
+        # outlives it. This is the only place a stop is allowed to block.
+        for stopping in self._stopping:
+            try:
+                stopping.process.communicate(timeout=0.25)
+            except subprocess.TimeoutExpired:
+                stopping.process.kill()
+                stopping.process.communicate()
+        self._stopping.clear()
 
     def _start_pending(self) -> None:
         while self._pending and len(self._active) < self.max_parallel:
@@ -722,17 +740,31 @@ class RemoteQueryPool:
             return None
         return _RemoteJob(entry, process, time.monotonic(), posix_fallback)
 
-    @staticmethod
-    def _stop_process(process: subprocess.Popen[bytes]) -> None:
-        if process.poll() is not None:
-            process.communicate()
-            return
-        process.terminate()
-        try:
-            process.communicate(timeout=0.25)
-        except subprocess.TimeoutExpired:
-            process.kill()
-            process.communicate()
+    def _stop_process(self, process: subprocess.Popen[bytes]) -> None:
+        """Ask a query to stop without blocking the UI thread.
+
+        Cancelling used to wait up to 250 ms per process; a refresh of six
+        hosts could freeze the popup for over a second. The process is reaped
+        on later polls and escalated to SIGKILL if it ignores SIGTERM.
+        """
+        if process.poll() is None:
+            process.terminate()
+        self._stopping.append(_StoppingProcess(process, time.monotonic()))
+
+    def _reap_stopping(self) -> None:
+        now = time.monotonic()
+        remaining: list[_StoppingProcess] = []
+        for stopping in self._stopping:
+            if stopping.process.poll() is not None:
+                for stream in (stopping.process.stdout, stopping.process.stderr):
+                    if stream is not None:
+                        stream.close()
+                continue
+            if not stopping.killed and now - stopping.since >= 0.25:
+                stopping.process.kill()
+                stopping.killed = True
+            remaining.append(stopping)
+        self._stopping = remaining
 
 
 def remote_control_command(

tests/test_remotes.py

@@ -282,6 +282,33 @@ class RemoteSshTests(unittest.TestCase):
 
         self.assertEqual(decode_remote_output("错误".encode("gbk")), "错误")
 
+    def test_cancel_does_not_block_and_reaps_the_process_later(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            slow_ssh = Path(directory) / "ssh"
+            # Ignore SIGTERM so the pool has to escalate to SIGKILL.
+            slow_ssh.write_text("#!/bin/sh\ntrap '' TERM\nsleep 5 & wait\n", encoding="utf-8")
+            slow_ssh.chmod(0o755)
+            pool = RemoteQueryPool(str(slow_ssh), query_timeout=5)
+            self.addCleanup(pool.close)
+            pool.enqueue(RemoteEntry("slow"))
+            time.sleep(0.05)
+            process = pool._active["slow"].process  # type: ignore[attr-defined]
+
+            started = time.monotonic()
+            pool.cancel("slow")
+            self.assertLess(time.monotonic() - started, 0.1)
+            self.assertFalse(pool.is_loading("slow"))
+
+            deadline = time.monotonic() + 3
+            while process.poll() is None and time.monotonic() < deadline:
+                pool.poll()
+                time.sleep(0.02)
+            self.assertIsNotNone(process.poll())
+            pool.poll()
+            self.assertEqual(pool._stopping, [])  # type: ignore[attr-defined]
+            assert process.stdout is not None
+            self.assertTrue(process.stdout.closed)
+
     def test_query_pool_times_out_without_blocking_other_results(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             slow_ssh = Path(directory) / "ssh"