fix(transport): eliminate startup hangs and endpoint hijacking in host-emulation IPC

.gitignore

@@ -3,6 +3,12 @@ build/
 Testing/
 CMakeUserPresets.json
 
+# Coverage counters. The Debug preset builds with CODE_COVERAGE, and every run
+# of an instrumented binary drops one of these wherever it was invoked from --
+# usually the repo root.
+*.profraw
+*.profdata
+
 # Editor / tooling
 .vscode/
 .cache/

py/host-emulator/src/host_emulator/emulator.py

@@ -6,14 +6,13 @@
 import json
 import logging
 import sys
-import time
-from pathlib import Path
-from threading import Thread
+from threading import Event, Thread
 from typing import Any, NoReturn
 
 import zmq
 
 from .common import UnhandledMessageError
+from .endpoint import EndpointLock, has_live_owner
 from .i2c import I2C
 from .pin import Pin, PinDirection, PinState
 from .uart import Uart
@@ -83,7 +82,12 @@ def __init__(
         self.i2cs = [self.i2c_1]
 
         self.emulator_thread = Thread(target=self.run)
-        self._ready = False
+        self._endpoint_lock = EndpointLock()
+        # Set once the bind phase has settled, either way. start() waits on
+        # this and then reads _startup_error, so a failure surfaces immediately
+        # with its real cause instead of as a timeout with a generic message.
+        self._bind_settled = Event()
+        self._startup_error: Exception | None = None
 
     def user_led1(self) -> Pin:
         return self.led_1
@@ -100,23 +104,49 @@ def uart1(self) -> Uart:
     def i2c1(self) -> I2C:
         return self.i2c_1
 
+    def _bind(self) -> None:
+        """Claim the receive endpoint and bind it.
+
+        No stale-file cleanup here, despite what the previous version did.
+        libzmq unlinks an ipc path before binding it, so a file left behind by a
+        killed process was never a problem -- and the unconditional unlink that
+        used to live here was itself the hazard: it would displace a *live*
+        emulator and take its endpoint, with no error on either side.
+
+        Two guards instead, in order. The lock is the guarantee: it is atomic,
+        so no other process using it can slip between the check and the bind.
+        The probe is the fallback for an owner that holds no lock.
+        """
+        endpoint = self.from_device_endpoint
+        if not self._endpoint_lock.try_acquire(endpoint):
+            msg = f"Endpoint {endpoint} is locked by another live process"
+            raise RuntimeError(msg)
+        if has_live_owner(endpoint):
+            msg = f"Endpoint {endpoint} is served by another live process"
+            raise RuntimeError(msg)
+
+        self.from_device_socket.bind(endpoint)
+        logger.debug("Bound to %s", endpoint)
+
     def run(self) -> None:
-        """Main emulator thread - BIND first, then signal ready."""
+        """Main emulator thread: bind, publish the outcome, then serve."""
         logger.debug("Starting emulator thread")
         try:
-            if self.from_device_endpoint.startswith("ipc://"):
-                socket_path = Path(self.from_device_endpoint.replace("ipc://", ""))
-                try:
-                    socket_path.unlink()
-                    logger.debug("Removed stale socket file: %s", socket_path)
-                except FileNotFoundError:
-                    pass
-
-            self.from_device_socket.bind(self.from_device_endpoint)
-            logger.debug("Bound to %s", self.from_device_endpoint)
+            self._bind()
+        except Exception as exc:  # Recorded here, re-raised by start().
+            self._startup_error = exc
+            logger.error("Emulator failed to bind: %s", exc)
+            self.from_device_socket.close()
+            self._endpoint_lock.release()
+            return
+        finally:
+            # Publish on every path out of the bind phase. start() is blocked
+            # on this; an unpublished outcome makes it wait out its whole
+            # timeout for something that will never arrive.
+            self._bind_settled.set()
 
+        try:
             self.running = True
-            self._ready = True
 
             while self.running:
                 try:
@@ -149,6 +179,9 @@ def run(self) -> None:
             logger.exception("Emulator thread error")
         finally:
             self.from_device_socket.close()
+            # Released only once the socket is closed, so the endpoint is never
+            # advertised as free while we still hold it.
+            self._endpoint_lock.release()
             logger.debug("Emulator thread exiting")
 
     def _handle_pin_message(self, json_message: dict[str, Any]) -> None:
@@ -176,20 +209,31 @@ def _handle_i2c_message(self, json_message: dict[str, Any]) -> None:
         raise UnhandledMessageError(f"I2C not found: {json_message.get('name')}")
 
     def start(self) -> None:
-        """Start emulator and wait until ready."""
+        """Start the emulator, raising if it could not claim its endpoint.
+
+        Waits for the bind outcome rather than for a duration, so the common
+        case returns as soon as the socket is bound and the failure case
+        reports why instead of timing out with a generic message.
+
+        Raises:
+            RuntimeError: If the endpoint is owned by another live process, or
+                the emulator thread never reported a bind outcome.
+        """
         self.emulator_thread.start()
 
-        timeout = 5.0
-        start_time = time.time()
-        while not self._ready:
-            if time.time() - start_time > timeout:
-                raise RuntimeError("Emulator failed to start within timeout")
-            time.sleep(0.01)
+        if not self._bind_settled.wait(timeout=5.0):
+            raise RuntimeError("Emulator thread never reported a bind outcome")
+        if self._startup_error is not None:
+            raise RuntimeError(
+                f"Emulator failed to start: {self._startup_error}"
+            ) from self._startup_error
 
         self.to_device_socket.connect(self.to_device_endpoint)
         logger.debug("Connected to %s", self.to_device_endpoint)
-
-        time.sleep(0.05)
+        # No settling sleep here. connect() is asynchronous and the device may
+        # not even have bound yet; libzmq retries in the background regardless.
+        # PAIR blocks rather than drops, and SNDTIMEO bounds the wait, so the
+        # sleep bought nothing. Test-side readiness is _wait_for_process_ready.
 
     def stop(self) -> None:
         """Stop emulator and clean up resources."""

py/host-emulator/src/host_emulator/endpoint.py

@@ -0,0 +1,128 @@
+"""Ownership guards for ipc:// endpoints.
+
+The Python counterpart of ``mcu::EndpointLock`` and
+``ZmqTransport::EndpointHasLiveOwner`` in ``src/libs/mcu/host/zmq_transport.cpp``.
+
+Both ends of the emulator IPC need this, for the same reason: libzmq unlinks an
+ipc path before binding it, unconditionally, and will happily displace a live
+listener and take its rendezvous name. Neither side sees an error -- the
+original owner keeps its existing connections, because the inode outlives the
+name, but every later connect() reaches the newcomer instead.
+
+The two implementations must agree on the lock file naming, or they do not
+exclude each other.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import logging
+import os
+import socket
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_IPC_SCHEME = "ipc://"
+_LOCK_SUFFIX = ".lock"
+_LOCK_MODE = 0o600
+
+
+def endpoint_path(endpoint: str) -> Path | None:
+    """Filesystem path an ipc:// endpoint binds to, or None for other transports."""
+    if not endpoint.startswith(_IPC_SCHEME):
+        return None
+    return Path(endpoint.removeprefix(_IPC_SCHEME))
+
+
+def has_live_owner(endpoint: str) -> bool:
+    """Whether another process is currently accepting on ``endpoint``.
+
+    libzmq's ipc:// transport is AF_UNIX/SOCK_STREAM, so a plain connect() is a
+    valid liveness probe with no ZMQ machinery involved: a path left behind by a
+    killed process refuses the connection, a live listener accepts it.
+
+    Every "cannot tell" answer is reported as live, so the caller never binds
+    over something it does not understand. Refusing to start is recoverable;
+    silently splitting the bus in two is not.
+
+    On its own this is racy -- another process can bind between the check and
+    the bind. :class:`EndpointLock` closes that window for anything using the
+    same lock; this remains the best available answer for an owner that is not.
+    """
+    path = endpoint_path(endpoint)
+    if path is None or not path.exists():
+        return False
+    if not path.is_socket():
+        logger.warning("Endpoint path exists and is not a socket: %s", path)
+        return True
+
+    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        probe.connect(str(path))
+    except ConnectionRefusedError:
+        return False  # Nobody listening: the owner is gone.
+    except OSError:
+        return True  # Cannot tell; assume live.
+    else:
+        return True  # Someone answered.
+    finally:
+        probe.close()
+
+
+class EndpointLock:
+    """Exclusive advisory ownership of a bind endpoint, held until released.
+
+    This is what makes "may I bind here" atomic. A liveness probe cannot be:
+    probing and binding are two separate calls, and another process can bind in
+    between. flock is arbitrated by the kernel, so that window does not exist.
+
+    It is also crash-safe, which an ``O_EXCL`` lock file is not: the lock lives
+    on the open file description and the kernel drops it when the fd closes --
+    including when the process dies -- so a killed run leaves nothing behind
+    that would block the next one.
+    """
+
+    def __init__(self) -> None:
+        self._fd: int | None = None
+
+    def try_acquire(self, endpoint: str) -> bool:
+        """Take the lock guarding ``endpoint``.
+
+        Returns False if another live process holds it. Endpoints with no
+        lockable path succeed trivially.
+        """
+        path = endpoint_path(endpoint)
+        if path is None:
+            return True  # No filesystem path to guard.
+
+        lock_path = path.parent / (path.name + _LOCK_SUFFIX)
+        try:
+            fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_CLOEXEC, _LOCK_MODE)
+        except OSError:
+            # Cannot lock here -- a read-only directory, for instance. Fall
+            # through to the liveness probe rather than refusing to start over
+            # a missing luxury.
+            logger.warning("Could not open lock file %s", lock_path)
+            return True
+
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            os.close(fd)
+            return False
+
+        self._fd = fd
+        return True
+
+    def release(self) -> None:
+        """Release the lock. Idempotent.
+
+        The lock file itself is deliberately left behind. Unlinking it would
+        reopen the race it exists to close: one process removing the file
+        another has already opened leaves the two holding locks on different
+        inodes, both believing they won.
+        """
+        if self._fd is not None:
+            os.close(self._fd)
+            self._fd = None

py/host-emulator/tests/conftest.py

@@ -51,16 +51,44 @@ def emulator() -> Generator[DeviceEmulator]:
             device_emulator.stop()
 
 
+def _endpoint_path(endpoint: str) -> Path | None:
+    """Filesystem path an ipc:// endpoint binds to, or None for other transports."""
+    if not endpoint.startswith("ipc://"):
+        return None
+    return Path(endpoint.removeprefix("ipc://"))
+
+
 def _wait_for_process_ready(
-    process: subprocess.Popen[bytes], timeout: float = 1.0
+    process: subprocess.Popen[bytes],
+    ready_path: Path | None,
+    timeout: float = 5.0,
+    poll_interval: float = 0.01,
 ) -> None:
-    """Wait for process to be running and responsive."""
-    start_time = time.time()
-    while time.time() - start_time < timeout:
+    """Block until the application has bound its receive endpoint.
+
+    Readiness is the appearance of the app's ipc socket file: the C++ transport
+    binds it inside ZmqTransport::Create(), before Create() returns.
+
+    This is narrower than "the app is ready". It does not prove the app finished
+    connecting to the emulator, nor that it reached its main loop -- both happen
+    after the bind and neither is observable from here. The per-test wait_for_*
+    helpers remain the real synchronisation for those.
+
+    The previous version had no success exit at all: it slept out its full
+    timeout on every call and treated "did not die" as ready.
+    """
+    if ready_path is None:
+        return  # No observable readiness signal for non-ipc endpoints.
+
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
         if process.poll() is not None:
             raise RuntimeError(f"Process exited with code {process.returncode}")
-        time.sleep(0.1)
-    time.sleep(0.1)
+        if ready_path.exists():
+            return
+        time.sleep(poll_interval)
+
+    raise RuntimeError(f"Process did not bind {ready_path} within {timeout}s")
 
 
 def _application_fixture_factory(option_name: str, display_name: str) -> Any:
@@ -89,14 +117,20 @@ def application_fixture(
             f"{display_name} executable not found: {app_executable}"
         )
 
+        # Clear any leftover socket file first, so its later appearance is
+        # evidence of *this* run binding rather than of a previous one.
+        ready_path = _endpoint_path(emulator.to_device_endpoint)
+        if ready_path is not None:
+            ready_path.unlink(missing_ok=True)
+
         app_process = subprocess.Popen(
             [str(app_executable)],
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
         )
 
         try:
-            _wait_for_process_ready(app_process)
+            _wait_for_process_ready(app_process, ready_path)
             yield app_process
 
         finally: