webgl: fix intermittent headless test hangs

cortex/export/headless.py

@@ -342,38 +342,64 @@ def headless_viewer(
     pw_thread = _PlaywrightThread()
 
     handle = None
+    # ------------------------------------------------------------------
+    # 3. Begin waiting for the WebSocket "connect" message on a *daemon*
+    #    thread *before* navigating, so we cannot miss it even if the browser
+    #    connects before page.goto() returns.
+    #
+    #    A daemon thread (rather than a ThreadPoolExecutor) is essential here:
+    #    server.get_client() blocks on a threading.Event with no timeout, so if
+    #    the browser never sends "connect" the getter parks forever. An
+    #    executor's worker is non-daemon and concurrent.futures joins every
+    #    worker at interpreter exit -- a parked getter would then wedge the
+    #    whole process at shutdown (observed as a multi-minute CI hang after
+    #    the tests have already finished). Daemon threads are never joined at
+    #    exit, so a stuck getter can never hold the process hostage.
+    # ------------------------------------------------------------------
+    connect_result: dict[str, Any] = {}
+
+    def _await_client() -> None:
+        try:
+            connect_result["handle"] = server.get_client()
+        except BaseException as exc:  # noqa: BLE001 - reported to main thread
+            connect_result["error"] = exc
+
+    client_thread = threading.Thread(
+        target=_await_client, name="headless-await-client", daemon=True
+    )
+
     try:
-        # ------------------------------------------------------------------
-        # 3. Begin waiting for the WebSocket "connect" message in a thread
-        #    *before* navigating, so we cannot miss it even if the browser
-        #    connects before page.goto() returns.
-        # ------------------------------------------------------------------
-        pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
-        fut = pool.submit(server.get_client)
+        client_thread.start()
         try:
             # Launch the browser and navigate.  python_interface.js runs on
             # load and sends "connect" over WebSocket, which unblocks
             # server.get_client().
             pw_thread.start(url, timeout=timeout)
 
-            # Retrieve the handle; it should already be ready by this point,
-            # but the timeout guard surfaces hung state clearly.
-            handle = fut.result(timeout=timeout)
+            # Wait (bounded) for the getter to return; the timeout guard
+            # surfaces hung state clearly instead of blocking indefinitely.
+            client_thread.join(timeout=timeout)
+            if "error" in connect_result:
+                raise connect_result["error"]
+            if "handle" not in connect_result:
+                raise TimeoutError(
+                    f"No WebSocket 'connect' received within {timeout:.0f}s"
+                )
+            handle = connect_result["handle"]
         except Exception as e:
-            fut.cancel()
             browser_errors = pw_thread.browser_errors
             detail = (
                 "\nBrowser errors:\n" + "\n".join(browser_errors)
                 if browser_errors
                 else "\nNo browser errors were captured."
             )
-            pool.shutdown(wait=False, cancel_futures=True)
+            # Unblock the (daemon) getter if it is still parked on the connect
+            # event so it exits promptly rather than lingering across sessions.
+            server.connect.set()
             raise RuntimeError(
                 f"Failed to establish WebSocket connection with headless browser at {url} "
                 f"within {timeout:.0f} seconds. {detail}"
             ) from e
-        else:
-            pool.shutdown(wait=False)
 
         assert not isinstance(handle, list)  # type narrowing to JSMixer
         handle.server = server

cortex/webgl/serve.py

@@ -28,6 +28,7 @@
 import tornado.ioloop
 import tornado.httpserver
 from tornado import websocket
+from tornado.netutil import bind_sockets
 from tornado.web import HTTPError
 
 cwd = os.path.split(os.path.abspath(__file__))[0]
@@ -310,9 +311,32 @@ def __init__(
             (r"/wsconnect/", ClientSocket, dict(parent=self)),
             (r"/(.*)", tornado.web.StaticFileHandler, dict(path=cwd)),
         ]
-        self.port = port
+        # Bind the listening socket(s) here, synchronously, in the calling
+        # thread -- rather than picking a random port and calling
+        # ``server.listen(port)`` later on the server thread. This fixes two
+        # long-standing sources of intermittent headless/CI hangs:
+        #   * ``port=0`` asks the OS for a guaranteed-free ephemeral port, so
+        #     we never collide with a lingering server (the old code used
+        #     ``random.randint`` and hoped for the best).
+        #   * a bind failure now raises *here*, visibly, in the caller. The old
+        #     ``listen()`` ran on the server thread, so an
+        #     "Address already in use" died silently and left callers to hang
+        #     for the full timeout waiting to connect to a server that had
+        #     never actually come up.
+        # The socket is bound immediately, so the OS accepts and queues client
+        # connections in the backlog even before the IOLoop starts serving --
+        # eliminating the connect race too.
+        self._sockets = bind_sockets(port if port is not None else 0)
+        # When port==0 the OS assigns the port; read the real value back so
+        # callers can build a correct URL.
+        self.port = self._sockets[0].getsockname()[1]
         self.response: Queue[Union[str, bytes]] = Queue()
         self.connect = threading.Event()
+        # Set by run() once self.server and self.ioloop exist and the server is
+        # about to serve. stop() waits on this so an early stop() (called before
+        # the server thread has finished starting up) neither hits missing
+        # attributes nor gets silently lost.
+        self._ready = threading.Event()
         self.sockets: list[websocket.WebSocketHandler] = []
 
     @property
@@ -333,13 +357,34 @@ def run(self):
             self.server = tornado.httpserver.HTTPServer(application, io_loop=ioloop)
         else:
             self.server = tornado.httpserver.HTTPServer(application)
-        self.server.listen(self.port)
+        # The socket(s) were already bound in __init__; just start serving on
+        # them. This cannot raise "Address already in use" on the server thread.
+        self.server.add_sockets(self._sockets)
+        # self.server and self.ioloop are now set; signal that stop() is safe.
+        self._ready.set()
         ioloop.start()
 
     def stop(self):
         print("Stopping server")
-        self.server.stop()
-        self.ioloop.stop()
+        # stop() may race a just-started server: run() sets self.ioloop and
+        # self.server on the server thread, so a stop() that arrives first would
+        # hit missing attributes -- or, worse, do nothing and let run() go on to
+        # serve forever. Wait for the server to finish coming up before stopping
+        # it.
+        if self.is_alive() and self._ready.wait(timeout=5):
+            self.server.stop()  # also closes the sockets it now owns
+            self.ioloop.stop()
+        else:
+            # The server thread never brought the server up (start() was never
+            # called, or it died during startup), so nothing owns the sockets
+            # bound in __init__. Close them here so we do not leak the
+            # listening port. is_alive() short-circuits the wait when start()
+            # was never called.
+            for sock in self._sockets:
+                try:
+                    sock.close()
+                except OSError:
+                    pass
 
     def send(self, **kwargs: Any) -> Union[list[JSON], list[None]]:
         msg = json.dumps(kwargs, cls=NPEncode, ensure_ascii=False)

cortex/webgl/view.py

@@ -930,7 +930,10 @@ def get_local_client(self):
             return JSMixer(self.srvsend, "window.viewer")
 
     if port is None:
-        port = random.randint(1024, 65536)
+        # Let the OS assign a guaranteed-free ephemeral port (WebApp binds
+        # port 0 and reads the real port back). This avoids the random-port
+        # collisions that made headless/CI runs intermittently hang.
+        port = 0
 
     server = WebApp([(r'/ctm/(.*)', CTMHandler),
                      (r'/data/(.*)', DataHandler),