test: pre-bind socket in run_uvicorn_in_thread, drop polling loop

The previous version polled server.started with time.sleep(0.001) until
uvicorn finished binding. We were waiting for uvicorn to tell us the
port — but we can just bind the socket ourselves and hand it to uvicorn
via server.run(sockets=[sock]).

Once sock.listen() returns, the kernel queues incoming connections (up
to the backlog). If a client connects before uvicorn's event loop
reaches accept(), the connection sits in the accept queue and is picked
up as soon as uvicorn is ready. The kernel is the synchronizer — no
cross-thread flag needed.

The port is now known before the thread starts. No polling, no sleep,
no wait at all.

Author: maxisbey

tests/test_helpers.py

@@ -9,60 +9,44 @@
 
 import uvicorn
 
-# How long to wait for the uvicorn server thread to reach `started`.
-# Generous to absorb CI scheduling delays — actual startup is typically <100ms.
-_SERVER_START_TIMEOUT_S = 20.0
 _SERVER_SHUTDOWN_TIMEOUT_S = 5.0
 
 
 @contextmanager
 def run_uvicorn_in_thread(app: Any, **config_kwargs: Any) -> Generator[str, None, None]:
-    """Run a uvicorn server in a background thread with an ephemeral port.
+    """Run a uvicorn server in a background thread on an ephemeral port.
 
-    This eliminates the TOCTOU race that occurs when a test picks a free port
-    with ``socket.bind((host, 0))``, releases it, then starts a server hoping
-    to rebind the same port — between release and rebind, another pytest-xdist
-    worker may claim it, causing connection errors or cross-test contamination.
+    The socket is bound and put into listening state *before* the thread
+    starts, so the port is known immediately with no wait. The kernel's
+    listen queue buffers any connections that arrive before uvicorn's event
+    loop reaches ``accept()``, so callers can connect as soon as this
+    function yields — no polling, no sleeps, no startup race.
 
-    With ``port=0``, the OS atomically assigns a free port at bind time; the
-    server holds it from that moment until shutdown. We read the actual port
-    back from uvicorn's bound socket after startup completes.
+    This also avoids the TOCTOU race of the old pick-a-port-then-rebind
+    pattern: the socket passed here is the one uvicorn serves on, with no
+    gap where another pytest-xdist worker could claim it.
 
     Args:
         app: ASGI application to serve.
         **config_kwargs: Additional keyword arguments for :class:`uvicorn.Config`
-            (e.g. ``log_level``, ``limit_concurrency``). ``host`` defaults to
-            ``127.0.0.1`` and ``port`` is forced to 0.
+            (e.g. ``log_level``). ``host``/``port`` are ignored since the
+            socket is pre-bound.
 
     Yields:
         The base URL of the running server, e.g. ``http://127.0.0.1:54321``.
-
-    Raises:
-        TimeoutError: If the server does not start within 20 seconds.
-        RuntimeError: If the server thread dies during startup.
     """
-    config_kwargs.setdefault("host", "127.0.0.1")
+    host = "127.0.0.1"
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind((host, 0))
+    sock.listen()
+    port = sock.getsockname()[1]
+
     config_kwargs.setdefault("log_level", "error")
-    config = uvicorn.Config(app=app, port=0, **config_kwargs)
-    server = uvicorn.Server(config=config)
+    server = uvicorn.Server(config=uvicorn.Config(app=app, **config_kwargs))
 
-    thread = threading.Thread(target=server.run, daemon=True)
+    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
     thread.start()
-
-    # uvicorn sets `server.started = True` at the end of `Server.startup()`,
-    # after sockets are bound and the lifespan startup phase has completed.
-    start = time.monotonic()
-    while not server.started:
-        if time.monotonic() - start > _SERVER_START_TIMEOUT_S:  # pragma: no cover
-            raise TimeoutError(f"uvicorn server failed to start within {_SERVER_START_TIMEOUT_S}s")
-        if not thread.is_alive():  # pragma: no cover
-            raise RuntimeError("uvicorn server thread exited during startup")
-        time.sleep(0.001)
-
-    # server.servers[0] is the asyncio.Server; its bound socket has the real port
-    port = server.servers[0].sockets[0].getsockname()[1]
-    host = config.host
-
     try:
         yield f"http://{host}:{port}"
     finally: