fix(e2e): stabilize listen-port races and reload polls

.agents/skills/e2e/SKILL.md

@@ -84,7 +84,9 @@ Keep URL template coverage split by resolver responsibility. Do not recreate a m
 - Import helpers only from `helpers`; update both imports and `__all__` in `helpers/__init__.py`
   when adding a helper.
 - Never hardcode ports. Use `find_free_port()`, `find_free_udp_port()`, or
-  `find_free_udp_port_pair()`.
+  `find_free_udp_port_pair()`. These allocate from a per-xdist-worker range
+  below typical ephemeral ports (14000-32399) so parallel client connections
+  cannot steal a listen port before rtp2httpd binds it.
 - Prefer module-scoped or class-scoped `R2HProcess` fixtures when tests share config and args.
 - Use per-test `R2HProcess` only for mutually exclusive configs, port-range behavior, timeout or
   log-capture cases, Unix socket cases, or tests whose process state must be isolated.
@@ -121,9 +123,11 @@ Port and process helpers:
 - `find_free_port()`
 - `find_free_udp_port()`
 - `find_free_udp_port_pair()`
+- `worker_port_range()`
 - `wait_for_port(port, host="127.0.0.1", timeout=5.0)`
 - `wait_for_unix_socket(path, timeout=5.0)`
 - `R2HProcess(binary, port, extra_args=None, config_content=None, capture_log=False, listen=None)`
+  (`start()` fail-fasts if the process exits and includes rtp2httpd logs)
 - `make_m3u_rtsp_config(port, rtsp_port, service_name="Test RTSP")`
 
 Config and file helpers:
@@ -242,7 +246,8 @@ assert_etag_cache_behavior("127.0.0.1", shared_r2h.port, "/playlist.m3u")
 
    - Missing marker or marker missing from `pyproject.toml`.
    - Helper added but not re-exported in `helpers/__init__.py`.
-   - Hardcoded port or port pair collision.
+   - Hardcoded port or port pair collision, or a listen port from `bind(0)`
+    in the ephemeral range (use `find_free_port()`).
    - Mock server started after rtp2httpd needs it.
    - `stream_get()` timeout too short for streaming or multicast.
    - External M3U fetch via `-M http://...` or `-M file://...` needs a short async wait.

e2e/helpers/__init__.py

@@ -6,7 +6,7 @@
 
 Sub-modules:
     constants   - project paths and platform constants
-    ports       - free port allocation and wait_for_port
+    ports       - per-worker listen-port allocation and wait_for_port
     http        - HTTP client helpers (http_get, http_request, stream_get)
     rtp         - RTP packet crafting and MulticastSender
     r2h_process - R2HProcess server wrapper
@@ -56,6 +56,7 @@
     ipv6_loopback_available,
     wait_for_port,
     wait_for_unix_socket,
+    worker_port_range,
 )
 from .r2h_process import R2HProcess, make_m3u_rtsp_config
 from .rtp import MulticastSender, make_rtp_packet
@@ -100,5 +101,6 @@
     "wait_for_port",
     "wait_for_status_payload",
     "wait_for_unix_socket",
+    "worker_port_range",
     "write_temp_file",
 ]

e2e/helpers/ports.py

@@ -1,75 +1,193 @@
-"""Port allocation helpers for E2E tests."""
+"""Port allocation helpers for E2E tests.
+
+``bind(0)`` draws from the kernel ephemeral range (Linux typically 32768-60999,
+macOS 49152-65535).  That races with pytest-xdist workers: a port returned by
+``bind(0)`` can be reused as another worker's client source port before
+rtp2httpd binds it.  CI failures such as ``rtp2httpd did not start on port
+49234`` were almost all macOS ephemeral ports.
+
+Listen ports are therefore taken from a per-xdist-worker range below those
+ephemeral windows.  A short cooldown avoids immediately recycling a port that
+may still be in TIME_WAIT after the previous process exits.
+"""
 
 from __future__ import annotations
 
+import os
 import socket
+import threading
 import time
+from collections import deque
 
+# Below Linux ip_local_port_range (32768+) and macOS net.inet.ip.portrange (49152+).
+_RANGE_BASE = 14000
+_RANGE_SIZE = 2300
+_MAX_RANGES = 8  # 14000-32399 even with more than 8 xdist workers
+_COOLDOWN = 64
+
+_lock = threading.Lock()
+_recent_tcp: deque[int] = deque(maxlen=_COOLDOWN)
+_recent_udp: deque[int] = deque(maxlen=_COOLDOWN)
+_tcp_cursor = 0
+_udp_cursor = 0
 
-def find_free_port(host: str = "127.0.0.1") -> int:
-    """Find a free TCP port on *host* (use "::1" for IPv6 loopback)."""
-    family = socket.AF_INET6 if ":" in host else socket.AF_INET
-    with socket.socket(family, socket.SOCK_STREAM) as s:
-        s.bind((host, 0))
-        return s.getsockname()[1]
+
+def xdist_worker_index() -> int:
+    """Return this process's pytest-xdist worker index (0 for serial runs)."""
+    worker = os.environ.get("PYTEST_XDIST_WORKER", "master")
+    if worker == "master":
+        return 0
+    if worker.startswith("gw") and worker[2:].isdigit():
+        return int(worker[2:]) + 1
+    return 0
+
+
+def worker_port_range() -> tuple[int, int]:
+    """Return the inclusive-exclusive TCP/UDP listen range for this worker."""
+    start = _RANGE_BASE + (xdist_worker_index() % _MAX_RANGES) * _RANGE_SIZE
+    return start, start + _RANGE_SIZE
 
 
 def ipv6_loopback_available() -> bool:
     """Return True when binding a TCP socket to ::1 works on this host."""
     try:
         with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
+            s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
             s.bind(("::1", 0))
         return True
     except OSError:
         return False
 
 
+def _tcp_probe_targets(host: str) -> list[tuple[int, str]]:
+    """Addresses whose bind success means rtp2httpd can listen here.
+
+    Tests usually wait on 127.0.0.1, but rtp2httpd binds 0.0.0.0 (and ::).
+    Probe the IPv4 wildcard so we do not hand out a port that is only free
+    on the loopback address.
+    """
+    if host in ("127.0.0.1", "0.0.0.0", ""):
+        return [(socket.AF_INET, "0.0.0.0")]
+    if ":" in host:
+        return [(socket.AF_INET6, host)]
+    return [(socket.AF_INET, host)]
+
+
+def _can_bind_tcp(host: str, port: int) -> bool:
+    sockets: list[socket.socket] = []
+    try:
+        for family, bind_host in _tcp_probe_targets(host):
+            sock = socket.socket(family, socket.SOCK_STREAM)
+            sockets.append(sock)
+            if family == socket.AF_INET6:
+                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
+            sock.bind((bind_host, port))
+        return True
+    except OSError:
+        return False
+    finally:
+        for sock in sockets:
+            sock.close()
+
+
+def _can_bind_udp(port: int) -> bool:
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+            sock.bind(("", port))
+        return True
+    except OSError:
+        return False
+
+
+def find_free_port(host: str = "127.0.0.1") -> int:
+    """Find a free TCP port on *host* (use "::1" for IPv6 loopback)."""
+    start, end = worker_port_range()
+    size = end - start
+    with _lock:
+        global _tcp_cursor
+        for step in range(size):
+            port = start + ((_tcp_cursor + step) % size)
+            if port in _recent_tcp:
+                continue
+            if _can_bind_tcp(host, port):
+                _tcp_cursor = (_tcp_cursor + step + 1) % size
+                _recent_tcp.append(port)
+                return port
+    raise RuntimeError(f"No free TCP port in {start}-{end - 1} for {host}")
+
+
 def find_free_udp_port() -> int:
     """Find a free UDP port."""
-    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
-        s.bind(("", 0))
-        return s.getsockname()[1]
+    start, end = worker_port_range()
+    size = end - start
+    with _lock:
+        global _udp_cursor
+        for step in range(size):
+            port = start + ((_udp_cursor + step) % size)
+            if port in _recent_udp:
+                continue
+            if _can_bind_udp(port):
+                _udp_cursor = (_udp_cursor + step + 1) % size
+                _recent_udp.append(port)
+                return port
+    raise RuntimeError(f"No free UDP port in {start}-{end - 1}")
 
 
 def find_free_udp_port_pair() -> tuple[int, int]:
     """Find a free even/odd UDP port pair (for RTP/RTCP)."""
-    for _ in range(100):
-        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s1:
-            s1.bind(("", 0))
-            p = s1.getsockname()[1]
-            if p % 2 != 0:
+    start, end = worker_port_range()
+    size = end - start
+    with _lock:
+        global _udp_cursor
+        for step in range(size):
+            port = start + ((_udp_cursor + step) % size)
+            if port % 2 != 0 or port + 1 >= end:
                 continue
-            try:
-                with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s2:
-                    s2.bind(("", p + 1))
-                    return p, p + 1
-            except OSError:
+            if port in _recent_udp or (port + 1) in _recent_udp:
                 continue
-    # fallback
-    return find_free_udp_port(), find_free_udp_port()
+            if _can_bind_udp(port) and _can_bind_udp(port + 1):
+                _udp_cursor = (_udp_cursor + step + 2) % size
+                _recent_udp.append(port)
+                _recent_udp.append(port + 1)
+                return port, port + 1
+    raise RuntimeError(f"No free UDP port pair in {start}-{end - 1}")
+
+
+def port_connectable(port: int, host: str = "127.0.0.1", timeout: float = 0.05) -> bool:
+    """Return True if *host:port* accepts a TCP connection right now."""
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            return True
+    except OSError:
+        return False
+
+
+def unix_socket_connectable(path: str, timeout: float = 0.05) -> bool:
+    """Return True if *path* accepts a Unix stream connection right now."""
+    try:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+            sock.settimeout(timeout)
+            sock.connect(path)
+        return True
+    except OSError:
+        return False
 
 
 def wait_for_port(port: int, host: str = "127.0.0.1", timeout: float = 5.0) -> bool:
     """Block until *port* is accepting TCP connections (or *timeout* expires)."""
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:
-        try:
-            with socket.create_connection((host, port), timeout=0.5):
-                return True
-        except TimeoutError, ConnectionRefusedError, OSError:
-            time.sleep(0.05)
+        if port_connectable(port, host=host, timeout=0.5):
+            return True
+        time.sleep(0.05)
     return False
 
 
 def wait_for_unix_socket(path: str, timeout: float = 5.0) -> bool:
     """Block until *path* is accepting Unix stream socket connections."""
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:
-        try:
-            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
-                sock.settimeout(0.5)
-                sock.connect(path)
-                return True
-        except OSError:
-            time.sleep(0.05)
+        if unix_socket_connectable(path, timeout=0.5):
+            return True
+        time.sleep(0.05)
     return False

e2e/helpers/r2h_process.py

@@ -5,9 +5,10 @@
 import os
 import subprocess
 import tempfile
+import time
 from pathlib import Path
 
-from .ports import wait_for_port, wait_for_unix_socket
+from .ports import port_connectable, unix_socket_connectable
 
 
 def make_m3u_rtsp_config(r2h_port: int, rtsp_port: int, channel_name: str, configured_url_query: str = "") -> str:
@@ -47,30 +48,18 @@ def __init__(
 
     def start(self, wait: bool = True) -> None:
         args = self._build_args()
-        if self.capture_log:
-            log_fd, self._log_path = tempfile.mkstemp(suffix=".log", prefix="r2h_log_")
-            self._log_handle = os.fdopen(log_fd, "w")
+        log_fd, self._log_path = tempfile.mkstemp(suffix=".log", prefix="r2h_log_")
+        self._log_handle = os.fdopen(log_fd, "w")
+        try:
             self.process = subprocess.Popen(args, stdout=self._log_handle, stderr=self._log_handle)
-        else:
-            self.process = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        except OSError as exc:
+            raise RuntimeError(f"failed to spawn rtp2httpd: {exc}.\nCommand: {' '.join(args)}") from exc
         if wait:
-            if self.wait_socket_path:
-                if not wait_for_unix_socket(self.wait_socket_path, timeout=6.0):
-                    self.stop()
-                    raise RuntimeError(
-                        "rtp2httpd did not start on Unix socket {}.\nCommand: {}".format(
-                            self.wait_socket_path, " ".join(args)
-                        )
-                    )
-            elif self.listen and self.listen.startswith("/"):
-                if not wait_for_unix_socket(self.listen, timeout=6.0):
-                    self.stop()
-                    raise RuntimeError(
-                        "rtp2httpd did not start on Unix socket {}.\nCommand: {}".format(self.listen, " ".join(args))
-                    )
-            elif self.port is not None and not wait_for_port(self.port, timeout=6.0):
+            error = self._wait_until_ready()
+            if error:
+                detail = self._startup_failure_detail(args, error)
                 self.stop()
-                raise RuntimeError(f"rtp2httpd did not start on port {self.port}.\nCommand: {' '.join(args)}")
+                raise RuntimeError(detail)
 
     def stop(self) -> None:
         if self.process and self.process.poll() is None:
@@ -98,12 +87,60 @@ def stop(self) -> None:
 
     def read_log(self) -> str:
         """Return the captured rtp2httpd stdout/stderr."""
-        assert self._log_path is not None, "read_log requires capture_log=True at construction time"
+        assert self._log_path is not None, "read_log requires the process to have been started"
+        if self._log_handle is not None:
+            self._log_handle.flush()
         with open(self._log_path) as f:
             return f.read()
 
     # -- internals -----------------------------------------------------------
 
+    def _ready_timeout(self) -> float:
+        return 10.0 if os.environ.get("CI") else 6.0
+
+    def _is_listening(self) -> bool:
+        if self.wait_socket_path:
+            return unix_socket_connectable(self.wait_socket_path)
+        if self.listen and self.listen.startswith("/"):
+            return unix_socket_connectable(self.listen)
+        if self.port is not None:
+            return port_connectable(self.port)
+        return True
+
+    def _wait_until_ready(self) -> str | None:
+        """Return None when ready, or a short reason if startup failed."""
+        assert self.process is not None
+        timeout = self._ready_timeout()
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if self.process.poll() is not None:
+                return f"process exited with code {self.process.returncode}"
+            if self._is_listening():
+                return None
+            time.sleep(0.05)
+        return f"timed out after {timeout:.1f}s waiting for listen"
+
+    def _startup_failure_detail(self, args: list[str], error: str) -> str:
+        if self.wait_socket_path:
+            target = f"Unix socket {self.wait_socket_path}"
+        elif self.listen and self.listen.startswith("/"):
+            target = f"Unix socket {self.listen}"
+        elif self.port is not None:
+            target = f"port {self.port}"
+        else:
+            target = "the configured listener"
+        log = ""
+        try:
+            log = self.read_log().strip()
+        except OSError:
+            pass
+        if len(log) > 4000:
+            log = log[-4000:]
+        detail = f"rtp2httpd did not start on {target}: {error}.\nCommand: {' '.join(args)}"
+        if log:
+            detail += f"\n--- rtp2httpd log ---\n{log}\n--- end log ---"
+        return detail
+
     def _build_args(self) -> list[str]:
         if self.config_content is not None:
             fd, path = tempfile.mkstemp(suffix=".conf", prefix="r2h_test_")