From 671ac945472cacebe05bc1d2633c3def180f13e4 Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 05:47:13 +0000 Subject: [PATCH 1/6] fix(transport): stop ZmqTransport hanging and lying about startup Create() could hang forever. The constructor waited on bind_cv_ with no deadline, so when the server thread failed to bind -- a bad path, a second instance -- the catch-all swallowed the throw, the thread exited, and the predicate could never become true. Bound that wait with startup_timeout and publish an outcome on every exit from the bind phase, including both catch blocks. A failed bind now surfaces as kOperationFailed from Create(). The constructor could also std::terminate. If connect() threw, the exception escaped while server_thread_ was joinable, and ~std::thread aborts the process before Create()'s handler runs. The constructor is now total: it never throws, never blocks past startup_timeout, and records failure in startup_error_ for Create() -- or a direct constructor caller -- to read via StartupStatus(). kConnected was set immediately after the asynchronous connect(), so it only ever meant "the constructor ran". WaitForConnection looped while kConnecting and therefore never iterated, leaving connect_timeout dead. Renamed the states to say what is actually known: kReady means our socket is bound and connect() was issued, and documents that it claims nothing about the peer. Peer liveness stays where it is genuinely observable -- the result of the Send you attempted. connect_timeout becomes startup_timeout with a real consumer; shutdown_timeout is deleted, as std::thread has no timed join and nothing could honour it. Also refuse to bind an endpoint a live process already serves. libzmq unlinks an ipc path before binding it, unconditionally, so it will silently displace a live listener and take its rendezvous name -- a second app instance, or a unit test run while the emulator is up, splits the bus in two with no error on either side. A connect(2) probe now detects that and fails startup instead. Stale files left by a killed process need no handling: the same unlink already makes them a non-issue. Co-Authored-By: Claude Opus 5 (1M context) --- src/libs/mcu/host/zmq_transport.cpp | 312 ++++++++++++++++++++-------- src/libs/mcu/host/zmq_transport.hpp | 79 +++++-- 2 files changed, 288 insertions(+), 103 deletions(-) diff --git a/src/libs/mcu/host/zmq_transport.cpp b/src/libs/mcu/host/zmq_transport.cpp index 5fc7bb3..93531d2 100644 --- a/src/libs/mcu/host/zmq_transport.cpp +++ b/src/libs/mcu/host/zmq_transport.cpp @@ -1,14 +1,79 @@ #include "zmq_transport.hpp" +#include +#include +#include + +#include #include -#include +#include +#include +#include +#include #include +#include #include #include "dispatcher.hpp" #include "libs/common/error.hpp" namespace mcu { +namespace { + +constexpr std::string_view kIpcScheme{"ipc://"}; + +// RAII for a bare file descriptor. The liveness probe below is the only place +// this file talks to POSIX sockets directly, and it must not leak an fd on any +// of its several early returns. +class FdGuard { + public: + explicit FdGuard(int descriptor) : fd_{descriptor} {} + FdGuard(const FdGuard&) = delete; + FdGuard(FdGuard&&) = delete; + auto operator=(const FdGuard&) -> FdGuard& = delete; + auto operator=(FdGuard&&) -> FdGuard& = delete; + ~FdGuard() { + if (fd_ >= 0) { + ::close(fd_); + } + } + + [[nodiscard]] auto Get() const -> int { return fd_; } + + private: + int fd_; +}; + +// True if some process is currently accepting on the AF_UNIX socket at `path`. +// +// libzmq's ipc:// transport is AF_UNIX/SOCK_STREAM, so a plain connect(2) 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 proceeds +// past something it does not understand. Refusing to start is recoverable; the +// hijack described in EndpointHasLiveOwner is not. +auto IpcPathHasLiveOwner(const std::string& path) -> bool { + sockaddr_un address{}; + address.sun_family = AF_UNIX; + if (path.size() >= sizeof(address.sun_path)) { + return true; // Too long to probe; assume live rather than guess. + } + path.copy(static_cast(address.sun_path), path.size()); + + const FdGuard probe{::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0)}; + if (probe.Get() < 0) { + return true; + } + + const auto* const address_ptr = reinterpret_cast(&address); + if (::connect(probe.Get(), address_ptr, sizeof(address)) == 0) { + return true; // Someone is listening -- hands off. + } + return errno != ECONNREFUSED; // ECONNREFUSED means the owner is gone. +} + +} // namespace auto ZmqTransport::Create(const std::string& to_emulator, const std::string& from_emulator, @@ -20,19 +85,15 @@ auto ZmqTransport::Create(const std::string& to_emulator, auto transport{std::make_unique(to_emulator, from_emulator, dispatcher, config)}; - // Wait for connection to establish - auto wait_result{transport->WaitForConnection(config.connect_timeout)}; - if (!wait_result) { - config.logger.Error("Connection timeout"); - return std::unexpected(wait_result.error()); + if (auto status{transport->StartupStatus()}; !status) { + config.logger.Error("ZmqTransport startup failed"); + return std::unexpected(status.error()); } config.logger.Info("ZmqTransport created successfully"); return transport; - } catch (const zmq::error_t& /*e*/) { - config.logger.Error("ZMQ error during creation"); - return std::unexpected(common::Error::kConnectionRefused); - } catch (...) { + } catch (...) { // NOLINT + // The constructor no longer throws, so this covers allocation failure only. config.logger.Error("Unknown error during creation"); return std::unexpected(common::Error::kUnknown); } @@ -45,26 +106,71 @@ ZmqTransport::ZmqTransport(const std::string& to_emulator, // NOLINT : config_{config}, dispatcher_{dispatcher} { LogDebug("Initializing ZmqTransport"); - SetSocketOptions(); + state_ = TransportState::kStarting; - state_ = TransportState::kConnecting; + // Nothing below may escape as an exception. Once server_thread_ is running, + // an exception leaving this constructor destroys a *joinable* std::thread, + // which calls std::terminate -- Create()'s handler never gets a look in. + try { + SetSocketOptions(); + + // Start server thread FIRST (it will BIND) + server_thread_ = + std::thread{&ZmqTransport::ServerThread, this, from_emulator}; + + switch (AwaitBind()) { + case BindOutcome::kPending: + FailStartup(common::Error::kTimeout, + "Startup timed out waiting for the server socket to bind"); + return; + case BindOutcome::kFailed: + FailStartup(common::Error::kOperationFailed, + "Startup failed: could not bind the server socket"); + return; + case BindOutcome::kBound: + break; + } - // Start server thread FIRST (it will BIND) - server_thread_ = - std::thread{&ZmqTransport::ServerThread, this, from_emulator}; + // Now CONNECT to emulator (emulator should already be bound) + LogDebug("Connecting to emulator"); + to_emulator_socket_.connect(to_emulator); - // Wait for server thread to complete bind before connecting - { - std::unique_lock lock(bind_mutex_); - bind_cv_.wait(lock, [this]() { return server_bound_.load(); }); + state_ = TransportState::kReady; + LogDebug("ZmqTransport ready (peer liveness unknown)"); + } catch (const zmq::error_t&) { + FailStartup(common::Error::kConnectionRefused, "Startup failed: ZMQ error"); + } catch (...) { // NOLINT + FailStartup(common::Error::kUnknown, "Startup failed: unknown error"); } +} + +auto ZmqTransport::FailStartup(common::Error error, + std::string_view msg) -> void { + startup_error_.store(error); + state_.store(TransportState::kFailed); + LogError(msg); +} - // Now CONNECT to emulator (emulator should already be bound) - LogDebug("Connecting to emulator"); - to_emulator_socket_.connect(to_emulator.c_str()); +auto ZmqTransport::AwaitBind() -> BindOutcome { + std::unique_lock lock(bind_mutex_); + // Bounded on purpose. If the server thread dies before it can bind -- a stale + // ipc file, a second instance already holding the endpoint -- bind_outcome_ + // stays kPending, and an unbounded wait here hangs the constructor forever. + bind_cv_.wait_for(lock, config_.startup_timeout, [this]() { + return bind_outcome_ != BindOutcome::kPending; + }); + return bind_outcome_; +} - state_ = TransportState::kConnected; - LogDebug("ZmqTransport initialized"); +auto ZmqTransport::SignalBind(BindOutcome outcome) -> void { + { + const std::lock_guard lock(bind_mutex_); + if (bind_outcome_ != BindOutcome::kPending) { + return; // First writer wins. + } + bind_outcome_ = outcome; + } + bind_cv_.notify_all(); } auto ZmqTransport::SetSocketOptions() -> void { @@ -78,22 +184,39 @@ auto ZmqTransport::SetSocketOptions() -> void { static_cast(config_.recv_timeout.count())); } -auto ZmqTransport::WaitForConnection(std::chrono::milliseconds timeout) - -> std::expected { - const auto deadline{std::chrono::steady_clock::now() + timeout}; - - while (state_ == TransportState::kConnecting) { - if (std::chrono::steady_clock::now() >= deadline) { - return std::unexpected(common::Error::kTimeout); - } - std::this_thread::sleep_for(std::chrono::milliseconds(10)); +// Whether another live process is already serving this endpoint. +// +// This is deliberately NOT stale-file cleanup. libzmq unlinks an ipc path +// before binding it, unconditionally, so a file left behind by a crashed +// process is already a non-problem -- bind() simply succeeds. +// +// The same unlink is what makes a *live* owner a problem. libzmq will happily +// remove a path another process is actively listening on and bind its own +// socket in place (verified: a second bind() to a held endpoint succeeds). +// Neither side sees an error. The original owner keeps its existing +// connections, because the inode outlives the name, but every subsequent +// connect() reaches the thief instead -- so a second app instance, or a unit +// test run while the emulator is up, silently splits the bus in two. +// +// libzmq gives us no way to ask it not to do that, so we check before handing +// it the endpoint and refuse to start rather than become the thief. +auto ZmqTransport::EndpointHasLiveOwner(const std::string& endpoint) const + -> bool { + const std::string_view endpoint_view{endpoint}; + if (!endpoint_view.starts_with(kIpcScheme)) { + return false; // Only ipc:// is probeable this way. } + const std::string path{endpoint_view.substr(kIpcScheme.size())}; - if (state_ == TransportState::kError) { - return std::unexpected(common::Error::kConnectionRefused); + std::error_code error{}; + if (!std::filesystem::exists(path, error) || error) { + return false; // Nothing there at all. } - - return {}; + if (!std::filesystem::is_socket(path, error) || error) { + LogWarning("Endpoint path exists and is not a socket"); + return true; // Not ours to reason about; do not bind over it. + } + return IpcPathHasLiveOwner(path); } ZmqTransport::~ZmqTransport() { @@ -128,8 +251,8 @@ ZmqTransport::~ZmqTransport() { auto ZmqTransport::Send(std::string_view data) -> std::expected { - if (state_ != TransportState::kConnected) { - LogWarning("Send failed: not connected"); + if (state_.load() != TransportState::kReady) { + LogWarning("Send failed: transport not ready"); return std::unexpected(common::Error::kInvalidState); } @@ -183,73 +306,86 @@ auto ZmqTransport::Send(std::string_view data) return std::unexpected(common::Error::kTimeout); } -void ZmqTransport::ServerThread(const std::string& endpoint) { - try { - LogDebug("ServerThread starting"); +auto ZmqTransport::ServerThread(const std::string& endpoint) -> void { + LogDebug("ServerThread starting"); + + zmq::socket_t socket{}; - zmq::socket_t socket{from_emulator_context_, zmq::socket_type::pair}; + // The bind phase. Every exit from this block -- fallthrough, early return, or + // exception -- must publish an outcome: the constructor is parked in + // AwaitBind(), and only a published outcome releases it before the timeout. + try { + socket = zmq::socket_t{from_emulator_context_, zmq::socket_type::pair}; socket.set(zmq::sockopt::linger, config_.linger_ms); socket.set(zmq::sockopt::rcvtimeo, static_cast(config_.poll_timeout.count())); + if (EndpointHasLiveOwner(endpoint)) { + LogError("Refusing to bind: endpoint is served by another live process"); + SignalBind(BindOutcome::kFailed); + return; + } socket.bind(endpoint); + } catch (const zmq::error_t&) { + LogError("ServerThread failed to bind"); + SignalBind(BindOutcome::kFailed); + return; + } catch (...) { // NOLINT + LogError("ServerThread failed to bind with an unknown error"); + SignalBind(BindOutcome::kFailed); + return; + } - // Signal that bind is complete - { - const std::lock_guard lock(bind_mutex_); - server_bound_ = true; - } - bind_cv_.notify_one(); + SignalBind(BindOutcome::kBound); + LogDebug("ServerThread bound and listening"); - LogDebug("ServerThread bound and listening"); + ServeLoop(socket); - while (running_) { - try { - zmq::message_t request{}; - auto result = socket.recv(request, zmq::recv_flags::none); + LogDebug("ServerThread exiting"); +} - if (!result) { - // Timeout or would block - check running flag - continue; - } +auto ZmqTransport::ServeLoop(zmq::socket_t& socket) -> void { + while (running_) { + try { + zmq::message_t request{}; + auto result = socket.recv(request, zmq::recv_flags::none); - const std::string_view request_str{ - static_cast(request.data()), request.size()}; - - auto response = dispatcher_.Dispatch(request.to_string()); - if (response) { - zmq::message_t reply{response.value().data(), - response.value().size()}; - socket.send(reply, zmq::send_flags::none); - } else { - LogWarning("Unhandled message in dispatcher"); - zmq::message_t reply{"Unhandled", 9}; - socket.send(reply, zmq::send_flags::none); - } + if (!result) { + // Timeout or would block - check running flag + continue; + } - } catch (const zmq::error_t& e) { - if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { - // Timeout - normal, check running flag - continue; - } - if (e.num() == ETERM) { - // Context terminated - time to exit - LogDebug("ServerThread received ETERM, exiting"); - break; - } - LogError("ServerThread ZMQ error"); + auto response = dispatcher_.Dispatch(request.to_string()); + if (response) { + zmq::message_t reply{response.value().data(), response.value().size()}; + socket.send(reply, zmq::send_flags::none); + } else { + LogWarning("Unhandled message in dispatcher"); + zmq::message_t reply{"Unhandled", 9}; + socket.send(reply, zmq::send_flags::none); } - } - LogDebug("ServerThread exiting"); - } catch (...) { // NOLINT - LogError("ServerThread caught exception"); + } catch (const zmq::error_t& e) { + if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { + // Timeout - normal, check running flag + continue; + } + if (e.num() == ETERM) { + // Context terminated - time to exit + LogDebug("ServeLoop received ETERM, exiting"); + return; + } + LogError("ServeLoop ZMQ error"); + } catch (...) { // NOLINT + LogError("ServeLoop caught exception"); + return; + } } } auto ZmqTransport::Receive() -> std::expected { - if (state_ != TransportState::kConnected) { - LogWarning("Receive failed: not connected"); + if (state_.load() != TransportState::kReady) { + LogWarning("Receive failed: transport not ready"); return std::unexpected(common::Error::kInvalidState); } diff --git a/src/libs/mcu/host/zmq_transport.hpp b/src/libs/mcu/host/zmq_transport.hpp index f52a3e4..1af43d2 100644 --- a/src/libs/mcu/host/zmq_transport.hpp +++ b/src/libs/mcu/host/zmq_transport.hpp @@ -1,7 +1,14 @@ #pragma once +#include +#include #include +#include #include +#include +#include +#include +#include #include #include @@ -12,11 +19,30 @@ namespace mcu { -enum class TransportState { - kDisconnected, - kConnecting, - kConnected, - kError, +// What the transport knows about ITSELF. +// +// No value here claims anything about the peer process. Both sockets are +// ZMQ_PAIR, and ZMQ offers no connection callback without a socket monitor, +// which this class deliberately does not use. Whether a peer is attached is +// only ever knowable from the result of the operation you just attempted -- +// which is why a missing emulator surfaces as Send() returning kTimeout rather +// than as a state value. +enum class TransportState : uint8_t { + // Constructed, startup not yet begun. Not observable from outside: the + // constructor moves to kStarting before any other thread holds a reference. + kUninitialized, + // Server thread launched; the bind outcome is not yet known. + kStarting, + // Our receive socket is bound and connect() has been issued on our send + // socket. The only state in which Send()/Receive() are permitted. + // + // This does NOT mean a peer is attached. connect() is asynchronous and + // returns before any peer exists, so a Send() in this state can still sit in + // ZMQ's mute state for send_timeout and come back as kTimeout. + kReady, + // Startup failed; terminal. Send()/Receive() return kInvalidState from here + // on, and StartupStatus() carries the reason. + kFailed, }; struct RetryConfig { @@ -27,8 +53,11 @@ struct RetryConfig { struct TransportConfig { std::chrono::milliseconds poll_timeout{50}; - std::chrono::milliseconds connect_timeout{5000}; - std::chrono::milliseconds shutdown_timeout{2000}; + // Bounds the one wait the constructor performs: the server thread's bind + // handshake. There is deliberately no connect timeout to go with it -- + // connect() on a PAIR socket is asynchronous and completes without a peer, + // so there would be nothing to wait for. + std::chrono::milliseconds startup_timeout{5000}; std::chrono::milliseconds send_timeout{1000}; std::chrono::milliseconds recv_timeout{5000}; int linger_ms{0}; // Discard pending messages on close @@ -62,25 +91,44 @@ class ZmqTransport : public Transport { -> std::expected override; auto Receive() -> std::expected override; - // New methods for connection management auto State() const -> TransportState { return state_.load(); } - auto IsConnected() const -> bool { - return state_.load() == TransportState::kConnected; + auto IsReady() const -> bool { + return state_.load() == TransportState::kReady; } - auto WaitForConnection(std::chrono::milliseconds timeout) - -> std::expected; + + // Outcome of the startup sequence the constructor ran. Never blocks: by the + // time the constructor has returned, startup has already succeeded, failed, + // or timed out. + auto StartupStatus() const -> std::expected { + const auto error{startup_error_.load()}; + if (error != common::Error::kOk) { + return std::unexpected(error); + } + return {}; + } + // Factory method - preferred way to create transport static auto Create(const std::string& to_emulator, const std::string& from_emulator, Dispatcher& dispatcher, const TransportConfig& config = {}) -> std::expected, common::Error>; - // Constructor - prefer using Create() factory method + // Never throws, and never blocks past config.startup_timeout. On failure the + // object is still fully constructed but permanently unusable: State() is + // kFailed, every Send()/Receive() returns kInvalidState, and StartupStatus() + // carries the reason. Prefer Create(), which makes that check for you. ZmqTransport(const std::string& to_emulator, const std::string& from_emulator, Dispatcher& dispatcher, const TransportConfig& config = {}); private: + enum class BindOutcome : uint8_t { kPending, kBound, kFailed }; + auto ServerThread(const std::string& endpoint) -> void; + auto ServeLoop(zmq::socket_t& socket) -> void; + auto SignalBind(BindOutcome outcome) -> void; + auto AwaitBind() -> BindOutcome; + auto FailStartup(common::Error error, std::string_view msg) -> void; + auto EndpointHasLiveOwner(const std::string& endpoint) const -> bool; auto SetSocketOptions() -> void; // Logging helpers to reduce cognitive complexity @@ -96,7 +144,8 @@ class ZmqTransport : public Transport { } TransportConfig config_; - std::atomic state_{TransportState::kDisconnected}; + std::atomic state_{TransportState::kUninitialized}; + std::atomic startup_error_{common::Error::kOk}; zmq::context_t to_emulator_context_{1}; zmq::socket_t to_emulator_socket_{to_emulator_context_, @@ -104,7 +153,7 @@ class ZmqTransport : public Transport { zmq::context_t from_emulator_context_{1}; std::atomic running_{true}; - std::atomic server_bound_{false}; + BindOutcome bind_outcome_{BindOutcome::kPending}; // guarded by bind_mutex_ std::condition_variable bind_cv_; std::mutex bind_mutex_; From ecc1e1ac1c0ea6c9e104968246f8b0ee9a8d37fb Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 05:53:47 +0000 Subject: [PATCH 2/6] test(transport): cover the startup failures that used to hang Adds test_zmq_transport_startup.cpp, in its own binary because its watchdog hard-exits the process on a regression and must not take other tests with it. Verified against the pre-change tree by reverting the production files: CreateFailsFastWhenBindEndpointIsUnbindable hangs (watchdog fires at 10s) CreateRefusesToStealEndpointFromLiveOwner fails (has_value() is true -- the hijack succeeded) CreateSucceedsOverStaleSocketFile passes The third is deliberately not a regression test, and says so: libzmq's own unlink always made stale files harmless. It guards the new liveness probe against the opposite mistake -- reading "file exists" as "owned" would refuse to start after any crash. The live-owner test asserts more than the error code. It sends a frame to the contested endpoint afterwards and requires the original owner's dispatcher to answer it, which is what actually proves the endpoint was not stolen. Also fixes three unchecked std::expected dereferences in the existing fixtures -- one bare deref and two value_or(nullptr) followed by a deref. They were latent before, when Create() could not fail; the startup fixes make them reachable, so they would have become null derefs rather than test failures. CTest gains TIMEOUT 60 on every unit test. A transport that used to wedge its constructor forever should fail CI, not stall it. Co-Authored-By: Claude Opus 5 (1M context) --- src/libs/mcu/host/CMakeLists.txt | 10 +- src/libs/mcu/host/test_host_i2c.cpp | 13 +- src/libs/mcu/host/test_host_uart.cpp | 13 +- src/libs/mcu/host/test_zmq_transport.cpp | 3 + .../mcu/host/test_zmq_transport_startup.cpp | 290 ++++++++++++++++++ 5 files changed, 316 insertions(+), 13 deletions(-) create mode 100644 src/libs/mcu/host/test_zmq_transport_startup.cpp diff --git a/src/libs/mcu/host/CMakeLists.txt b/src/libs/mcu/host/CMakeLists.txt index d88b337..87908e0 100644 --- a/src/libs/mcu/host/CMakeLists.txt +++ b/src/libs/mcu/host/CMakeLists.txt @@ -24,7 +24,10 @@ function(add_host_unit_test name source) add_executable(${name} ${source}) target_compile_options(${name} PRIVATE ${COMMON_COMPILE_OPTIONS}) target_link_libraries(${name} PRIVATE GTest::GTest ${ARGN}) - gtest_discover_tests(${name} PROPERTIES LABELS "unit") + # TIMEOUT is a backstop, not a budget: unit tests here run in well under a + # second, and a startup bug in the transport used to wedge the constructor + # forever rather than fail. A hung test should fail CI, not stall it. + gtest_discover_tests(${name} PROPERTIES LABELS "unit" TIMEOUT 60) if(CODE_COVERAGE) # Exclusions are inherited from global add_code_coverage_all_targets() target_code_coverage(${name} AUTO ALL) @@ -33,6 +36,11 @@ endfunction() # cppzmq is needed because zmq_transport.hpp includes zmq.hpp add_host_unit_test(test_host_transport test_zmq_transport.cpp host_transport cppzmq) +# Separate binary from test_host_transport on purpose: these tests bound a +# potential hang with a watchdog that hard-exits the process, which must not +# take unrelated tests down with it. +add_host_unit_test(test_host_transport_startup test_zmq_transport_startup.cpp + host_transport cppzmq) add_host_unit_test(test_messages test_messages.cpp nlohmann_json::nlohmann_json) add_host_unit_test(test_dispatcher test_dispatcher.cpp) add_host_unit_test(test_host_uart test_host_uart.cpp diff --git a/src/libs/mcu/host/test_host_i2c.cpp b/src/libs/mcu/host/test_host_i2c.cpp index 1ce3f35..dc76617 100644 --- a/src/libs/mcu/host/test_host_i2c.cpp +++ b/src/libs/mcu/host/test_host_i2c.cpp @@ -35,12 +35,13 @@ class HostI2CTest : public ::testing::Test { // later) dispatcher_ = std::make_unique(receiver_map_storage_); - // Create transport - device_transport_ = - mcu::ZmqTransport::Create("ipc:///tmp/test_i2c_device_emulator.ipc", - "ipc:///tmp/test_i2c_emulator_device.ipc", - *dispatcher_) - .value_or(nullptr); + // Create transport. Assert rather than value_or(nullptr): Create can fail, + // and a null transport is dereferenced two lines down. + auto transport_result = mcu::ZmqTransport::Create( + "ipc:///tmp/test_i2c_device_emulator.ipc", + "ipc:///tmp/test_i2c_emulator_device.ipc", *dispatcher_); + ASSERT_TRUE(transport_result.has_value()); + device_transport_ = std::move(transport_result.value()); // Now create I2C with transport i2c_ = diff --git a/src/libs/mcu/host/test_host_uart.cpp b/src/libs/mcu/host/test_host_uart.cpp index 8de589e..1e3e2fa 100644 --- a/src/libs/mcu/host/test_host_uart.cpp +++ b/src/libs/mcu/host/test_host_uart.cpp @@ -33,12 +33,13 @@ class HostUartTest : public ::testing::Test { // later) dispatcher_ = std::make_unique(receiver_map_storage_); - // Create transport - device_transport_ = - mcu::ZmqTransport::Create("ipc:///tmp/test_uart_device_emulator.ipc", - "ipc:///tmp/test_uart_emulator_device.ipc", - *dispatcher_) - .value_or(nullptr); + // Create transport. Assert rather than value_or(nullptr): Create can fail, + // and a null transport is dereferenced two lines down. + auto transport_result = mcu::ZmqTransport::Create( + "ipc:///tmp/test_uart_device_emulator.ipc", + "ipc:///tmp/test_uart_emulator_device.ipc", *dispatcher_); + ASSERT_TRUE(transport_result.has_value()); + device_transport_ = std::move(transport_result.value()); // Now create UART with transport uart_ = std::make_unique("UART 1", *device_transport_); diff --git a/src/libs/mcu/host/test_zmq_transport.cpp b/src/libs/mcu/host/test_zmq_transport.cpp index 5bfc67e..d9e20de 100644 --- a/src/libs/mcu/host/test_zmq_transport.cpp +++ b/src/libs/mcu/host/test_zmq_transport.cpp @@ -81,6 +81,9 @@ TEST_F(ZmqTransportTest, SendReceive) { auto transport = mcu::ZmqTransport::Create("ipc:///tmp/device_emulator.ipc", "ipc:///tmp/emulator_device.ipc", dispatcher); + // has_value() rather than the expected itself: std::expected's operator bool + // is explicit, so gtest's AssertionResult will not take it. + ASSERT_TRUE(transport.has_value()); auto result = (*transport)->Send("Hello"); ASSERT_TRUE(result); auto response = (*transport)->Receive(); diff --git a/src/libs/mcu/host/test_zmq_transport_startup.cpp b/src/libs/mcu/host/test_zmq_transport_startup.cpp new file mode 100644 index 0000000..32f6390 --- /dev/null +++ b/src/libs/mcu/host/test_zmq_transport_startup.cpp @@ -0,0 +1,290 @@ +// Startup-failure tests for ZmqTransport. +// +// Every test here targets a path that used to hang the constructor forever +// rather than return: bind_cv_.wait() had no deadline, so any failure that +// stopped the server thread before it published a bind outcome parked the +// caller permanently. They live in their own binary because the watchdog below +// hard-exits the process on a regression, which must not take unrelated tests +// with it. + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "libs/common/error.hpp" +#include "libs/common/logger.hpp" +#include "libs/mcu/host/dispatcher.hpp" +#include "libs/mcu/host/receiver.hpp" +#include "libs/mcu/host/zmq_transport.hpp" + +namespace { + +using CreateResult = + std::expected, common::Error>; + +constexpr auto kWatchdogBudget = std::chrono::seconds{10}; +constexpr auto kStartupTimeout = std::chrono::milliseconds{2000}; + +// Collects log output so a test can assert on the reason for a failure, not +// merely that one occurred. The mutex is load-bearing: the bind phase logs from +// the server thread while the test thread is still inside the constructor. +class RecordingLogger : public common::Logger { + public: + auto Debug(std::string_view msg) -> void override { Record(msg); } + auto Info(std::string_view msg) -> void override { Record(msg); } + auto Warning(std::string_view msg) -> void override { Record(msg); } + auto Error(std::string_view msg) -> void override { Record(msg); } + + auto Contains(std::string_view needle) const -> bool { + const std::lock_guard lock(mutex_); + return std::ranges::any_of(messages_, [needle](const std::string& msg) { + return msg.find(needle) != std::string::npos; + }); + } + + private: + auto Record(std::string_view msg) -> void { + const std::lock_guard lock(mutex_); + messages_.emplace_back(msg); + } + + mutable std::mutex mutex_; + std::vector messages_; +}; + +// Answers anything, so a test can prove a message reached a given transport's +// dispatcher rather than some other process that stole the endpoint. +class EchoReceiver : public mcu::Receiver { + public: + auto Receive(const std::string_view& message) + -> std::expected override { + return std::string{"echo:"} + std::string{message}; + } +}; + +// Endpoints are per-test and per-process so a stuck or slow test can never +// collide with another, and never with the emulator's real endpoints. +auto UniqueEndpoint(std::string_view suffix) -> std::string { + const auto* const info = + ::testing::UnitTest::GetInstance()->current_test_info(); + return std::string{"ipc:///tmp/zt_startup_"} + info->name() + "_" + + std::to_string(::getpid()) + "_" + std::string{suffix} + ".ipc"; +} + +auto PathOf(const std::string& endpoint) -> std::string { + return endpoint.substr(std::string_view{"ipc://"}.size()); +} + +auto MakeConfig(common::Logger& logger) -> mcu::TransportConfig { + // TransportConfig has user-provided constructors, so it is not an aggregate: + // designated initialisers will not compile. Assign after construction. + mcu::TransportConfig config{logger}; + config.startup_timeout = kStartupTimeout; + return config; +} + +// Runs `fn` with a hard time budget. +// +// The hang being guarded against is inside a constructor called on this thread, +// so bounding it needs a second thread. std::async is unusable: its future's +// destructor joins, so a stuck task would hang the test at scope exit anyway. +// A detached thread over a shared_ptr-owned packaged_task lets the stuck thread +// outlive the call without dangling. +auto RunWithWatchdog(std::function action) + -> std::optional { + auto task = + std::make_shared>(std::move(action)); + auto future = task->get_future(); + std::thread{[task]() { (*task)(); }}.detach(); + + if (future.wait_for(kWatchdogBudget) != std::future_status::ready) { + return std::nullopt; + } + return future.get(); +} + +// Sends one message to `endpoint` from a fresh peer and returns whatever came +// back, or nullopt if the exchange did not complete. Extracted from the test +// body to keep its cognitive complexity under the clang-tidy threshold. +auto RoundTripThroughEndpoint(const std::string& endpoint) + -> std::optional { + zmq::context_t context{1}; + zmq::socket_t peer{context, zmq::socket_type::pair}; + peer.set(zmq::sockopt::linger, 0); + peer.set(zmq::sockopt::sndtimeo, 2000); + peer.set(zmq::sockopt::rcvtimeo, 2000); + peer.connect(endpoint); + + std::optional reply_text{}; + if (peer.send(zmq::str_buffer("ping"), zmq::send_flags::none)) { + zmq::message_t reply{}; + if (peer.recv(reply, zmq::recv_flags::none)) { + reply_text = reply.to_string(); + } + } + + peer.close(); + context.close(); + return reply_text; +} + +// A wedged thread cannot be unwound, and letting it linger through static +// destruction of live ZMQ contexts crashes unpredictably. Exit loudly instead: +// ctest sees a non-zero status, and the ADD_FAILURE line names the regression. +[[noreturn]] auto AbortOnHang() -> void { + ADD_FAILURE() << "ZmqTransport::Create() never returned within " + << kWatchdogBudget.count() + << "s -- the startup hang has regressed"; + std::cout << std::flush; + std::_Exit(EXIT_FAILURE); +} + +// Leaves behind exactly what a SIGKILLed process leaves: a bound socket file +// with no listener. Closing without unlinking is the whole point. +auto LeaveStaleSocketFile(const std::string& path) -> void { + sockaddr_un address{}; + address.sun_family = AF_UNIX; + ASSERT_LT(path.size(), sizeof(address.sun_path)); + path.copy(static_cast(address.sun_path), path.size()); + + const int descriptor = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + ASSERT_GE(descriptor, 0); + const auto* const address_ptr = reinterpret_cast(&address); + ASSERT_EQ(::bind(descriptor, address_ptr, sizeof(address)), 0); + ::close(descriptor); +} + +class ZmqTransportStartupTest : public ::testing::Test { + protected: + void TearDown() override { + std::error_code error{}; + for (const auto& endpoint : cleanup_) { + std::filesystem::remove(PathOf(endpoint), error); + } + } + + auto TrackForCleanup(const std::string& endpoint) -> std::string { + cleanup_.push_back(endpoint); + return endpoint; + } + + RecordingLogger logger_; + const mcu::ReceiverMap empty_receivers_; + + private: + std::vector cleanup_; +}; + +// A bind that cannot succeed must be reported, not waited out. The elapsed +// assertion is the real subject: a kOperationFailed that took startup_timeout +// would mean the outcome was never published and we merely timed out. +TEST_F(ZmqTransportStartupTest, CreateFailsFastWhenBindEndpointIsUnbindable) { + mcu::Dispatcher dispatcher{empty_receivers_}; + auto config = MakeConfig(logger_); + const auto to_endpoint = TrackForCleanup(UniqueEndpoint("to")); + const std::string from_endpoint{ + "ipc:///tmp/zt_startup_no_such_directory/from.ipc"}; + + const auto start = std::chrono::steady_clock::now(); + const auto outcome = RunWithWatchdog([&]() { + return mcu::ZmqTransport::Create(to_endpoint, from_endpoint, dispatcher, + config); + }); + const auto elapsed = std::chrono::steady_clock::now() - start; + + if (!outcome) { + AbortOnHang(); + } + ASSERT_FALSE(outcome->has_value()); + EXPECT_EQ(outcome->error(), common::Error::kOperationFailed); + EXPECT_LT(elapsed, kStartupTimeout) + << "reported rather than timed out is the point"; + EXPECT_TRUE(logger_.Contains("ServerThread failed to bind")); +} + +// libzmq unlinks an ipc path before binding it, so it will displace a live +// listener and take its name with no error on either side. Refusing to start is +// the only way to keep the first owner's endpoint intact. +TEST_F(ZmqTransportStartupTest, CreateRefusesToStealEndpointFromLiveOwner) { + EchoReceiver echo; + const mcu::ReceiverMap receivers{ + {[](const std::string_view&) { return true; }, std::ref(echo)}}; + mcu::Dispatcher owner_dispatcher{receivers}; + auto owner_config = MakeConfig(logger_); + + const auto owner_to = TrackForCleanup(UniqueEndpoint("owner_to")); + const auto contested = TrackForCleanup(UniqueEndpoint("contested")); + + auto owner = mcu::ZmqTransport::Create(owner_to, contested, owner_dispatcher, + owner_config); + ASSERT_TRUE(owner.has_value()); + + RecordingLogger thief_logger; + mcu::Dispatcher thief_dispatcher{empty_receivers_}; + auto thief_config = MakeConfig(thief_logger); + const auto thief_to = TrackForCleanup(UniqueEndpoint("thief_to")); + + const auto outcome = RunWithWatchdog([&]() { + return mcu::ZmqTransport::Create(thief_to, contested, thief_dispatcher, + thief_config); + }); + + if (!outcome) { + AbortOnHang(); + } + ASSERT_FALSE(outcome->has_value()); + EXPECT_EQ(outcome->error(), common::Error::kOperationFailed); + EXPECT_TRUE(thief_logger.Contains("served by another live process")); + + // The assertion that actually proves no hijack occurred: a fresh peer + // connecting to the contested endpoint still reaches the original owner. + EXPECT_EQ(RoundTripThroughEndpoint(contested), + std::optional{"echo:ping"}); +} + +// The mirror image, and unlike its two neighbours this is NOT a regression +// test -- it passes on the pre-change code too, because libzmq's own unlink +// always made stale files a non-problem. It guards the liveness probe added +// alongside it: an over-eager probe that read "file exists" as "owned" would +// refuse to start after any crash, turning a harmless leftover into a failure. +TEST_F(ZmqTransportStartupTest, CreateSucceedsOverStaleSocketFile) { + mcu::Dispatcher dispatcher{empty_receivers_}; + auto config = MakeConfig(logger_); + const auto to_endpoint = TrackForCleanup(UniqueEndpoint("to")); + const auto from_endpoint = TrackForCleanup(UniqueEndpoint("from")); + + LeaveStaleSocketFile(PathOf(from_endpoint)); + ASSERT_TRUE(std::filesystem::exists(PathOf(from_endpoint))); + + const auto outcome = RunWithWatchdog([&]() { + return mcu::ZmqTransport::Create(to_endpoint, from_endpoint, dispatcher, + config); + }); + + if (!outcome) { + AbortOnHang(); + } + ASSERT_TRUE(outcome->has_value()); + EXPECT_TRUE(outcome->value()->IsReady()); +} + +} // namespace From 73c045b3b17f3dde3679f7a61a025ed4c61ea885 Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 06:01:04 +0000 Subject: [PATCH 3/6] test: synchronise fixtures on conditions instead of sleeps Every fixture waited out a fixed duration where an ordering constraint or an observable condition was available. The C++ fixtures each slept 100ms hoping their emulator thread had bound, then another 100ms hoping the transport had connected. The first is now an ordering guarantee: the socket is a member bound on the test thread before the serving thread is created, so the transport's connect() happens-after the bind by thread creation alone. The second is a probe Send -- on a PAIR socket a send succeeds only once a pipe to the peer exists, so success IS the readiness signal, and the emulator loops already skip anything that fails to decode. test_zmq_transport's teardown slept 100ms before terminating its context. That sleep was load-bearing: the serving loop has no handler, and a context terminated under zmq::poll throws ETERM out of the thread, which is std::terminate. Stopping the loop and joining first removes the sleep and the hazard together. The 50ms "connect time" sleep before the unsolicited-data send was never what made that test work -- the socket had no SNDTIMEO, so its send already blocked until the pipe came up. It now has bounded timeouts and asserts the result. The 50ms "give handler time to execute" sleep was likewise unnecessary: HostUart::Receive runs the handler before building the ack, so a reply in hand already happens-after the handler. But handler_called and received_data were a genuine data race across the server thread, sleep or no sleep, and are now an atomic and a mutex-guarded vector. _wait_for_process_ready had no success exit at all: it slept out its full timeout on every call and treated "did not die in 1.1s" as ready. It now waits for the app's ipc socket file to appear, which the transport binds inside Create(), and raises on timeout instead of falling through. The docstring is explicit that this proves the bind and nothing further. Verified both branches: a process that exits early and one that runs without ever binding each fail with a specific message. Test endpoints no longer collide with the emulator's real ones -- the transport fixture had been binding ipc:///tmp/device_emulator.ipc, byte-identical to HostBoard::Endpoints and DeviceEmulator's defaults. Total test time drops from 10.2s to 4.1s. 50 consecutive unit runs are green. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/host_emulator/emulator.py | 6 +- py/host-emulator/tests/conftest.py | 48 ++++++++-- src/libs/mcu/host/test_host_i2c.cpp | 28 +++--- src/libs/mcu/host/test_host_uart.cpp | 74 ++++++++++----- src/libs/mcu/host/test_zmq_transport.cpp | 90 +++++++++++-------- 5 files changed, 164 insertions(+), 82 deletions(-) diff --git a/py/host-emulator/src/host_emulator/emulator.py b/py/host-emulator/src/host_emulator/emulator.py index 2ce253d..67bef32 100755 --- a/py/host-emulator/src/host_emulator/emulator.py +++ b/py/host-emulator/src/host_emulator/emulator.py @@ -188,8 +188,10 @@ def start(self) -> None: 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.""" diff --git a/py/host-emulator/tests/conftest.py b/py/host-emulator/tests/conftest.py index fa39e57..c56576c 100644 --- a/py/host-emulator/tests/conftest.py +++ b/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,6 +117,12 @@ 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, @@ -96,7 +130,7 @@ def application_fixture( ) try: - _wait_for_process_ready(app_process) + _wait_for_process_ready(app_process, ready_path) yield app_process finally: diff --git a/src/libs/mcu/host/test_host_i2c.cpp b/src/libs/mcu/host/test_host_i2c.cpp index dc76617..bacb827 100644 --- a/src/libs/mcu/host/test_host_i2c.cpp +++ b/src/libs/mcu/host/test_host_i2c.cpp @@ -24,13 +24,15 @@ class HostI2CTest : public ::testing::Test { } void SetUp() override { - // Start emulator thread + // Bind on the test thread, before the emulator thread exists, so the + // transport's connect() below happens-after the bind by thread creation + // alone. This replaces a 100ms sleep that only made the race unlikely. + emulator_socket_.set(zmq::sockopt::linger, 0); + emulator_socket_.bind("ipc:///tmp/test_i2c_device_emulator.ipc"); + emulator_running_ = true; emulator_thread_ = std::thread{[this]() { EmulatorLoop(); }}; - // Give emulator time to start - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - // Create dispatcher with empty receiver map (will update via reference // later) dispatcher_ = std::make_unique(receiver_map_storage_); @@ -50,8 +52,12 @@ class HostI2CTest : public ::testing::Test { // Add I2C to receiver map (dispatcher holds reference, so this updates it) receiver_map_storage_.emplace_back(IsJson, std::ref(*i2c_)); - // Give transport time to connect - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + // Wait for the condition rather than for a duration. On a PAIR socket a + // send succeeds only once a pipe to the peer exists, so a successful probe + // IS the readiness signal, and connect latency is absorbed by SNDTIMEO. + // The emulator loop skips anything that fails to decode, so this non-JSON + // probe is swallowed with no reply and needs no protocol support. + ASSERT_TRUE(device_transport_->Send("probe")); } void TearDown() override { @@ -59,12 +65,14 @@ class HostI2CTest : public ::testing::Test { device_transport_.reset(); dispatcher_.reset(); + // Stop and join before closing anything the thread is using: terminating a + // context out from under a running loop throws ETERM inside it. emulator_running_ = false; if (emulator_thread_.joinable()) { - emulator_context_.shutdown(); - emulator_context_.close(); emulator_thread_.join(); } + emulator_socket_.close(); + emulator_context_.close(); } void EmulatorLoop() { @@ -72,8 +80,7 @@ class HostI2CTest : public ::testing::Test { std::map> i2c_device_buffers; try { - zmq::socket_t socket{emulator_context_, zmq::socket_type::pair}; - socket.bind("ipc:///tmp/test_i2c_device_emulator.ipc"); + zmq::socket_t& socket = emulator_socket_; while (emulator_running_) { std::array items = { @@ -148,6 +155,7 @@ class HostI2CTest : public ::testing::Test { std::unique_ptr device_transport_; std::unique_ptr i2c_; zmq::context_t emulator_context_{1}; + zmq::socket_t emulator_socket_{emulator_context_, zmq::socket_type::pair}; std::thread emulator_thread_; std::atomic emulator_running_{false}; }; diff --git a/src/libs/mcu/host/test_host_uart.cpp b/src/libs/mcu/host/test_host_uart.cpp index 1e3e2fa..b532193 100644 --- a/src/libs/mcu/host/test_host_uart.cpp +++ b/src/libs/mcu/host/test_host_uart.cpp @@ -1,9 +1,11 @@ #include #include +#include #include #include #include +#include #include #include #include @@ -22,13 +24,15 @@ class HostUartTest : public ::testing::Test { } void SetUp() override { - // Start emulator thread + // Bind on the test thread, before the emulator thread exists, so the + // transport's connect() below happens-after the bind by thread creation + // alone. This replaces a 100ms sleep that only made the race unlikely. + emulator_socket_.set(zmq::sockopt::linger, 0); + emulator_socket_.bind("ipc:///tmp/test_uart_device_emulator.ipc"); + emulator_running_ = true; emulator_thread_ = std::thread{[this]() { EmulatorLoop(); }}; - // Give emulator time to start - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - // Create dispatcher with empty receiver map (will update via reference // later) dispatcher_ = std::make_unique(receiver_map_storage_); @@ -47,8 +51,12 @@ class HostUartTest : public ::testing::Test { // Add UART to receiver map (dispatcher holds reference, so this updates it) receiver_map_storage_.emplace_back(IsJson, std::ref(*uart_)); - // Give transport time to connect - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + // Wait for the condition rather than for a duration. On a PAIR socket a + // send succeeds only once a pipe to the peer exists, so a successful probe + // IS the readiness signal, and connect latency is absorbed by SNDTIMEO. + // The emulator loop skips anything that fails to decode, so this non-JSON + // probe is swallowed with no reply and needs no protocol support. + ASSERT_TRUE(device_transport_->Send("probe")); } void TearDown() override { @@ -56,22 +64,22 @@ class HostUartTest : public ::testing::Test { device_transport_.reset(); dispatcher_.reset(); + // Stop and join before closing anything the thread is using: terminating a + // context out from under a running loop throws ETERM inside it. emulator_running_ = false; if (emulator_thread_.joinable()) { - emulator_context_.shutdown(); - emulator_context_.close(); - unsolicited_context_.shutdown(); - unsolicited_context_.close(); emulator_thread_.join(); } + emulator_socket_.close(); + emulator_context_.close(); + unsolicited_context_.close(); } void EmulatorLoop() { std::vector uart_rx_buffer; try { - zmq::socket_t socket{emulator_context_, zmq::socket_type::pair}; - socket.bind("ipc:///tmp/test_uart_device_emulator.ipc"); + zmq::socket_t& socket = emulator_socket_; while (emulator_running_) { std::array items = { @@ -148,6 +156,7 @@ class HostUartTest : public ::testing::Test { std::unique_ptr device_transport_; std::unique_ptr uart_; zmq::context_t emulator_context_{1}; + zmq::socket_t emulator_socket_{emulator_context_, zmq::socket_type::pair}; zmq::context_t unsolicited_context_{1}; std::thread emulator_thread_; std::atomic emulator_running_{false}; @@ -243,14 +252,22 @@ TEST_F(HostUartTest, RxHandlerUnsolicitedData) { auto init_result = uart_->Init(config); ASSERT_TRUE(init_result); - // Track received data via handler + // Track received data via handler. The handler runs on the transport's + // server thread while this thread reads the results, so both need + // synchronisation -- a plain bool and vector here were a data race + // regardless of any sleep. std::vector received_data{}; - bool handler_called{false}; + std::mutex received_mutex{}; + std::atomic handler_called{false}; // Register RxHandler - auto handler_result = uart_->SetRxHandler( - [&received_data, &handler_called](const std::byte* data, size_t size) { - received_data.assign(data, data + size); + auto handler_result = + uart_->SetRxHandler([&received_data, &received_mutex, &handler_called]( + const std::byte* data, size_t size) { + { + const std::lock_guard lock(received_mutex); + received_data.assign(data, data + size); + } handler_called = true; }); ASSERT_TRUE(handler_result); @@ -272,23 +289,32 @@ TEST_F(HostUartTest, RxHandlerUnsolicitedData) { // arrival) zmq::socket_t unsolicited_socket{unsolicited_context_, zmq::socket_type::pair}; + // Timeouts instead of a "connect time" sleep. This socket had none, so its + // send already blocked until the pipe came up -- the sleep was never what + // made this work, it just hid an unbounded wait behind a bounded-looking one. + unsolicited_socket.set(zmq::sockopt::linger, 0); + unsolicited_socket.set(zmq::sockopt::sndtimeo, 2000); + unsolicited_socket.set(zmq::sockopt::rcvtimeo, 2000); unsolicited_socket.connect("ipc:///tmp/test_uart_emulator_device.ipc"); - std::this_thread::sleep_for(std::chrono::milliseconds(50)); // Connect time const auto request_str = mcu::Encode(unsolicited_request); - unsolicited_socket.send(zmq::buffer(request_str), zmq::send_flags::none); - - // Wait for response (dispatcher should route and UART should respond) + ASSERT_TRUE( + unsolicited_socket.send(zmq::buffer(request_str), zmq::send_flags::none)); + + // Wait for response (dispatcher should route and UART should respond). + // + // This recv is also the handler barrier, which is why no sleep follows it: + // HostUart::Receive invokes rx_handler_ before it builds the ack, and the + // server thread only replies once Dispatch has returned. A reply in hand + // therefore happens-after the handler ran. zmq::message_t response_msg{}; const auto recv_result{ unsolicited_socket.recv(response_msg, zmq::recv_flags::none)}; ASSERT_TRUE(recv_result.has_value()); - // Give handler time to execute - std::this_thread::sleep_for(std::chrono::milliseconds(50)); - // Verify handler was called with correct data EXPECT_TRUE(handler_called); + const std::lock_guard lock(received_mutex); EXPECT_EQ(received_data, test_data); } diff --git a/src/libs/mcu/host/test_zmq_transport.cpp b/src/libs/mcu/host/test_zmq_transport.cpp index d9e20de..a75cc8b 100644 --- a/src/libs/mcu/host/test_zmq_transport.cpp +++ b/src/libs/mcu/host/test_zmq_transport.cpp @@ -1,7 +1,11 @@ #include +#include +#include +#include #include #include +#include #include "dispatcher.hpp" #include "zmq_transport.hpp" @@ -9,68 +13,75 @@ namespace mcu { namespace { +// Deliberately not the emulator's real endpoints. This fixture used to bind +// ipc:///tmp/device_emulator.ipc -- byte-identical to HostBoard::Endpoints and +// to DeviceEmulator's defaults -- so running the unit tests while an emulator +// or blinky was up had them fighting over the same paths. +constexpr auto kEmulatorEndpoint = + "ipc:///tmp/test_transport_device_emulator.ipc"; +constexpr auto kDeviceEndpoint = + "ipc:///tmp/test_transport_emulator_device.ipc"; + class ZmqTransportTest : public ::testing::Test { protected: void SetUp() override { - server_thread_ = std::thread{&ZmqTransportTest::ServerThread, this, - "ipc:///tmp/device_emulator.ipc"}; + // Bind on the test thread, before the serving thread exists. Create() in + // the test body then happens-after this bind by thread creation alone. + // Previously the fixture launched a thread that bound the endpoint and the + // test called Create() immediately, racing it -- nothing ordered the two, + // and only ZMQ's connect retry hid the race. + socket_.set(zmq::sockopt::linger, 0); + socket_.bind(kEmulatorEndpoint); + server_thread_ = std::thread{[this]() { ServerLoop(); }}; } void TearDown() override { - try { - running_ = false; - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - - context_.shutdown(); - context_.close(); - + // Order matters. The old teardown terminated the context first and slept + // 100ms hoping the thread would notice; a context terminated out from under + // zmq::poll throws ETERM straight out of a thread with no handler, which is + // std::terminate. Stopping the loop and joining first removes both the + // sleep and the hazard. The socket must close before its context, or + // context teardown blocks waiting for it. + running_ = false; + if (server_thread_.joinable()) { server_thread_.join(); - } catch (const zmq::error_t& e) { - // Shutting down - if (e.num() != ETERM) { - std::cout << "Error: " << e.what() << '\n'; - } - } catch (const std::exception& e) { - std::cout << "Error: " << e.what() << '\n'; - } catch (...) { - std::cout << "Unknown error" << '\n'; } + socket_.close(); + context_.close(); } private: - void ServerThread(const std::string& endpoint) { - zmq::socket_t socket{context_, zmq::socket_type::pair}; - socket.bind(endpoint); + void ServerLoop() { while (running_) { std::array items = { - {{.socket = static_cast(socket), + {{.socket = static_cast(socket_), .fd = 0, .events = ZMQ_POLLIN, .revents = 0}}}; + // Bounded, so the loop notices running_ within one interval and the + // join above never waits long. const int ret{zmq::poll(items.data(), 1, std::chrono::milliseconds{50})}; - if (ret == 0) { - // Timeout occurred, check the stop condition - if (!running_) { - break; - } - } else if (ret > 0) { - zmq::message_t request{}; - if (socket.recv(request, zmq::recv_flags::none)) { - const std::string_view request_str{ - static_cast(request.data()), request.size()}; - if (request_str == "Hello") { - socket.send(zmq::str_buffer("World"), zmq::send_flags::none); - } - } else { - socket.send(zmq::str_buffer("Unknown"), zmq::send_flags::none); + if (ret <= 0) { + continue; + } + + zmq::message_t request{}; + if (socket_.recv(request, zmq::recv_flags::none)) { + const std::string_view request_str{ + static_cast(request.data()), request.size()}; + if (request_str == "Hello") { + socket_.send(zmq::str_buffer("World"), zmq::send_flags::none); } + } else { + socket_.send(zmq::str_buffer("Unknown"), zmq::send_flags::none); } } } zmq::context_t context_{1}; + zmq::socket_t socket_{context_, zmq::socket_type::pair}; std::thread server_thread_; std::atomic running_{true}; }; @@ -79,11 +90,12 @@ TEST_F(ZmqTransportTest, SendReceive) { const ReceiverMap receiver_map{}; Dispatcher dispatcher{receiver_map}; auto transport = - mcu::ZmqTransport::Create("ipc:///tmp/device_emulator.ipc", - "ipc:///tmp/emulator_device.ipc", dispatcher); + mcu::ZmqTransport::Create(kEmulatorEndpoint, kDeviceEndpoint, dispatcher); // has_value() rather than the expected itself: std::expected's operator bool // is explicit, so gtest's AssertionResult will not take it. ASSERT_TRUE(transport.has_value()); + EXPECT_TRUE((*transport)->IsReady()); + auto result = (*transport)->Send("Hello"); ASSERT_TRUE(result); auto response = (*transport)->Receive(); From 76c23e46ab2e7245ba4903b25b57b3e2905eb974 Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 06:27:49 +0000 Subject: [PATCH 4/6] fix(transport): close the bind race with an flock'd endpoint lock The connect(2) liveness probe added earlier narrowed the endpoint-hijack window but could not close it: probing and binding are two syscalls, and another process can bind in between. libzmq then unlinks whichever socket file it finds, so both processes end up believing they own the endpoint and neither sees an error. Measured, with 12 processes released together by a shared spin barrier so the window is genuinely contended: the probe alone yields between 1 and 7 winners per run. flock yields exactly 1, every run. flock is the right primitive here for two reasons. It is arbitrated by the kernel, so there is no window to lose. And the lock lives on the open file description, so the kernel drops it when the fd closes -- including on process death -- which means a SIGKILLed run leaves nothing stale behind. An O_EXCL lock file would have needed its own liveness check, recreating the problem one level up. The lock file itself is deliberately never unlinked: removing it would let two processes hold locks on different inodes and both proceed. The probe stays, behind the lock, as the answer for an owner that holds no lock -- an older build, or anything else listening on that path. ConcurrentBindsProduceExactlyOneOwner makes the measurement permanent. It forks twelve contenders through the same barrier and requires exactly one to succeed; with the lock disabled it reports 4-5 winners. Test fixtures now derive their endpoints from the pid. gtest_discover_tests gives each case its own process, so the fixed paths meant `ctest -j` cases contended for one endpoint -- silently corrupting each other before this change and failing loudly after. ctest -j8 was already red before any of this work; it now passes repeatedly. Fixtures also remove their own lock files, which the transport cannot do for itself. Co-Authored-By: Claude Opus 5 (1M context) --- src/libs/mcu/host/test_host_i2c.cpp | 30 +++++- src/libs/mcu/host/test_host_uart.cpp | 32 +++++- src/libs/mcu/host/test_zmq_transport.cpp | 45 ++++++--- .../mcu/host/test_zmq_transport_startup.cpp | 97 ++++++++++++++++++- src/libs/mcu/host/zmq_transport.cpp | 53 ++++++++++ src/libs/mcu/host/zmq_transport.hpp | 32 ++++++ 6 files changed, 270 insertions(+), 19 deletions(-) diff --git a/src/libs/mcu/host/test_host_i2c.cpp b/src/libs/mcu/host/test_host_i2c.cpp index bacb827..299d39c 100644 --- a/src/libs/mcu/host/test_host_i2c.cpp +++ b/src/libs/mcu/host/test_host_i2c.cpp @@ -1,12 +1,15 @@ #include +#include #include #include #include #include #include +#include #include #include +#include #include #include @@ -23,12 +26,20 @@ class HostI2CTest : public ::testing::Test { return message.starts_with("{") && message.ends_with("}"); } + // Per-process endpoints. gtest_discover_tests gives every case its own + // process, so a fixed path made `ctest -j` cases contend for one endpoint -- + // silently corrupting each other before EndpointLock, loudly after. + static auto Endpoint(std::string_view role) -> std::string { + return "ipc:///tmp/test_i2c_" + std::string{role} + "_" + + std::to_string(::getpid()) + ".ipc"; + } + void SetUp() override { // Bind on the test thread, before the emulator thread exists, so the // transport's connect() below happens-after the bind by thread creation // alone. This replaces a 100ms sleep that only made the race unlikely. emulator_socket_.set(zmq::sockopt::linger, 0); - emulator_socket_.bind("ipc:///tmp/test_i2c_device_emulator.ipc"); + emulator_socket_.bind(device_emulator_endpoint_); emulator_running_ = true; emulator_thread_ = std::thread{[this]() { EmulatorLoop(); }}; @@ -40,8 +51,7 @@ class HostI2CTest : public ::testing::Test { // Create transport. Assert rather than value_or(nullptr): Create can fail, // and a null transport is dereferenced two lines down. auto transport_result = mcu::ZmqTransport::Create( - "ipc:///tmp/test_i2c_device_emulator.ipc", - "ipc:///tmp/test_i2c_emulator_device.ipc", *dispatcher_); + device_emulator_endpoint_, emulator_device_endpoint_, *dispatcher_); ASSERT_TRUE(transport_result.has_value()); device_transport_ = std::move(transport_result.value()); @@ -73,6 +83,18 @@ class HostI2CTest : public ::testing::Test { } emulator_socket_.close(); emulator_context_.close(); + + // The transport never unlinks its own lock file -- doing so would reopen + // the race it closes -- so per-process test endpoints would otherwise pile + // up in /tmp, one pair per test case per run. + std::error_code error{}; + for (const auto& endpoint : + {device_emulator_endpoint_, emulator_device_endpoint_}) { + const std::string path{ + endpoint.substr(std::string_view{"ipc://"}.size())}; + std::filesystem::remove(path, error); + std::filesystem::remove(path + ".lock", error); + } } void EmulatorLoop() { @@ -150,6 +172,8 @@ class HostI2CTest : public ::testing::Test { } } + const std::string device_emulator_endpoint_{Endpoint("device_emulator")}; + const std::string emulator_device_endpoint_{Endpoint("emulator_device")}; mcu::ReceiverMap receiver_map_storage_; std::unique_ptr dispatcher_; std::unique_ptr device_transport_; diff --git a/src/libs/mcu/host/test_host_uart.cpp b/src/libs/mcu/host/test_host_uart.cpp index b532193..4d06d86 100644 --- a/src/libs/mcu/host/test_host_uart.cpp +++ b/src/libs/mcu/host/test_host_uart.cpp @@ -1,12 +1,15 @@ #include +#include #include #include #include #include #include +#include #include #include +#include #include #include @@ -23,12 +26,20 @@ class HostUartTest : public ::testing::Test { return message.starts_with("{") && message.ends_with("}"); } + // Per-process endpoints. gtest_discover_tests gives every case its own + // process, so a fixed path made `ctest -j` cases contend for one endpoint -- + // silently corrupting each other before EndpointLock, loudly after. + static auto Endpoint(std::string_view role) -> std::string { + return "ipc:///tmp/test_uart_" + std::string{role} + "_" + + std::to_string(::getpid()) + ".ipc"; + } + void SetUp() override { // Bind on the test thread, before the emulator thread exists, so the // transport's connect() below happens-after the bind by thread creation // alone. This replaces a 100ms sleep that only made the race unlikely. emulator_socket_.set(zmq::sockopt::linger, 0); - emulator_socket_.bind("ipc:///tmp/test_uart_device_emulator.ipc"); + emulator_socket_.bind(device_emulator_endpoint_); emulator_running_ = true; emulator_thread_ = std::thread{[this]() { EmulatorLoop(); }}; @@ -40,8 +51,7 @@ class HostUartTest : public ::testing::Test { // Create transport. Assert rather than value_or(nullptr): Create can fail, // and a null transport is dereferenced two lines down. auto transport_result = mcu::ZmqTransport::Create( - "ipc:///tmp/test_uart_device_emulator.ipc", - "ipc:///tmp/test_uart_emulator_device.ipc", *dispatcher_); + device_emulator_endpoint_, emulator_device_endpoint_, *dispatcher_); ASSERT_TRUE(transport_result.has_value()); device_transport_ = std::move(transport_result.value()); @@ -73,6 +83,18 @@ class HostUartTest : public ::testing::Test { emulator_socket_.close(); emulator_context_.close(); unsolicited_context_.close(); + + // The transport never unlinks its own lock file -- doing so would reopen + // the race it closes -- so per-process test endpoints would otherwise pile + // up in /tmp, one pair per test case per run. + std::error_code error{}; + for (const auto& endpoint : + {device_emulator_endpoint_, emulator_device_endpoint_}) { + const std::string path{ + endpoint.substr(std::string_view{"ipc://"}.size())}; + std::filesystem::remove(path, error); + std::filesystem::remove(path + ".lock", error); + } } void EmulatorLoop() { @@ -151,6 +173,8 @@ class HostUartTest : public ::testing::Test { } } + const std::string device_emulator_endpoint_{Endpoint("device_emulator")}; + const std::string emulator_device_endpoint_{Endpoint("emulator_device")}; mcu::ReceiverMap receiver_map_storage_; std::unique_ptr dispatcher_; std::unique_ptr device_transport_; @@ -295,7 +319,7 @@ TEST_F(HostUartTest, RxHandlerUnsolicitedData) { unsolicited_socket.set(zmq::sockopt::linger, 0); unsolicited_socket.set(zmq::sockopt::sndtimeo, 2000); unsolicited_socket.set(zmq::sockopt::rcvtimeo, 2000); - unsolicited_socket.connect("ipc:///tmp/test_uart_emulator_device.ipc"); + unsolicited_socket.connect(emulator_device_endpoint_); const auto request_str = mcu::Encode(unsolicited_request); ASSERT_TRUE( diff --git a/src/libs/mcu/host/test_zmq_transport.cpp b/src/libs/mcu/host/test_zmq_transport.cpp index a75cc8b..cd818d9 100644 --- a/src/libs/mcu/host/test_zmq_transport.cpp +++ b/src/libs/mcu/host/test_zmq_transport.cpp @@ -1,9 +1,13 @@ #include +#include #include #include #include +#include #include +#include +#include #include #include @@ -13,14 +17,18 @@ namespace mcu { namespace { -// Deliberately not the emulator's real endpoints. This fixture used to bind -// ipc:///tmp/device_emulator.ipc -- byte-identical to HostBoard::Endpoints and -// to DeviceEmulator's defaults -- so running the unit tests while an emulator -// or blinky was up had them fighting over the same paths. -constexpr auto kEmulatorEndpoint = - "ipc:///tmp/test_transport_device_emulator.ipc"; -constexpr auto kDeviceEndpoint = - "ipc:///tmp/test_transport_emulator_device.ipc"; +// Deliberately not the emulator's real endpoints, and per-process. +// +// This fixture used to bind ipc:///tmp/device_emulator.ipc -- byte-identical to +// HostBoard::Endpoints and to DeviceEmulator's defaults -- so running the unit +// tests while an emulator or blinky was up had them fighting over one path. The +// pid suffix additionally lets `ctest -j` work: gtest_discover_tests gives each +// case its own process, and with a fixed path those processes contended for the +// same endpoint. +auto Endpoint(std::string_view role) -> std::string { + return "ipc:///tmp/test_transport_" + std::string{role} + "_" + + std::to_string(::getpid()) + ".ipc"; +} class ZmqTransportTest : public ::testing::Test { protected: @@ -31,7 +39,7 @@ class ZmqTransportTest : public ::testing::Test { // test called Create() immediately, racing it -- nothing ordered the two, // and only ZMQ's connect retry hid the race. socket_.set(zmq::sockopt::linger, 0); - socket_.bind(kEmulatorEndpoint); + socket_.bind(emulator_endpoint_); server_thread_ = std::thread{[this]() { ServerLoop(); }}; } @@ -48,6 +56,16 @@ class ZmqTransportTest : public ::testing::Test { } socket_.close(); context_.close(); + + // The transport never unlinks its own lock file (that would reopen the race + // it closes), so clean up this process's endpoints here. + std::error_code error{}; + for (const auto& endpoint : {emulator_endpoint_, device_endpoint_}) { + const std::string path{ + endpoint.substr(std::string_view{"ipc://"}.size())}; + std::filesystem::remove(path, error); + std::filesystem::remove(path + ".lock", error); + } } private: @@ -80,6 +98,11 @@ class ZmqTransportTest : public ::testing::Test { } } + protected: + const std::string emulator_endpoint_{Endpoint("device_emulator")}; + const std::string device_endpoint_{Endpoint("emulator_device")}; + + private: zmq::context_t context_{1}; zmq::socket_t socket_{context_, zmq::socket_type::pair}; std::thread server_thread_; @@ -89,8 +112,8 @@ class ZmqTransportTest : public ::testing::Test { TEST_F(ZmqTransportTest, SendReceive) { const ReceiverMap receiver_map{}; Dispatcher dispatcher{receiver_map}; - auto transport = - mcu::ZmqTransport::Create(kEmulatorEndpoint, kDeviceEndpoint, dispatcher); + auto transport = mcu::ZmqTransport::Create(emulator_endpoint_, + device_endpoint_, dispatcher); // has_value() rather than the expected itself: std::expected's operator bool // is explicit, so gtest's AssertionResult will not take it. ASSERT_TRUE(transport.has_value()); diff --git a/src/libs/mcu/host/test_zmq_transport_startup.cpp b/src/libs/mcu/host/test_zmq_transport_startup.cpp index 32f6390..58159ce 100644 --- a/src/libs/mcu/host/test_zmq_transport_startup.cpp +++ b/src/libs/mcu/host/test_zmq_transport_startup.cpp @@ -8,11 +8,14 @@ // with it. #include +#include #include #include +#include #include #include +#include #include #include #include @@ -173,12 +176,82 @@ auto LeaveStaleSocketFile(const std::string& path) -> void { ::close(descriptor); } +// Forks kContenders processes that all attempt to bind `contested` at the same +// instant and returns how many believed they succeeded. +// +// The shared spin barrier is the point. Launching processes normally spreads +// their arrivals over milliseconds, so the first one binds and the rest see a +// live owner -- the window never opens and the race looks absent. Releasing +// them together from one store contends it properly. +// +// Lives outside the test body to keep TestBody's cognitive complexity under the +// clang-tidy threshold. +auto CountConcurrentBindWinners(const std::string& contested) -> int { + constexpr int kContenders = 12; + + auto* gate = static_cast*>( + ::mmap(nullptr, sizeof(std::atomic), PROT_READ | PROT_WRITE, + MAP_SHARED | MAP_ANONYMOUS, -1, 0)); + if (gate == MAP_FAILED) { + return -1; + } + gate->store(0); + + for (int i = 0; i < kContenders; ++i) { + if (::fork() != 0) { + continue; // Parent. + } + // Child. Exits via _exit so it never runs gtest teardown or atexit handlers + // belonging to the parent's test process. + common::NullLogger logger; + mcu::TransportConfig config{logger}; + config.startup_timeout = kStartupTimeout; + const mcu::ReceiverMap receivers{}; + mcu::Dispatcher dispatcher{receivers}; + const std::string own = contested + ".peer" + std::to_string(i); + + while (gate->load(std::memory_order_acquire) == 0) { + // Spin rather than sleep: the point is to arrive together. + } + auto transport = + mcu::ZmqTransport::Create(own, contested, dispatcher, config); + const bool won = transport.has_value(); + if (won) { + // Hold it briefly so later arrivals genuinely contend. + std::this_thread::sleep_for(std::chrono::milliseconds{400}); + } + ::_exit(won ? 0 : 1); + } + + std::this_thread::sleep_for(std::chrono::milliseconds{250}); // all spinning + gate->store(1, std::memory_order_release); // release + + int winners = 0; + for (int i = 0; i < kContenders; ++i) { + int status = 0; + ::wait(&status); + if (WIFEXITED(status) && WEXITSTATUS(status) == 0) { + ++winners; + } + } + ::munmap(gate, sizeof(std::atomic)); + + std::error_code error{}; + for (int i = 0; i < kContenders; ++i) { + const std::string own = PathOf(contested) + ".peer" + std::to_string(i); + std::filesystem::remove(own, error); + std::filesystem::remove(own + ".lock", error); + } + return winners; +} + class ZmqTransportStartupTest : public ::testing::Test { protected: void TearDown() override { std::error_code error{}; for (const auto& endpoint : cleanup_) { std::filesystem::remove(PathOf(endpoint), error); + std::filesystem::remove(PathOf(endpoint) + ".lock", error); } } @@ -253,7 +326,10 @@ TEST_F(ZmqTransportStartupTest, CreateRefusesToStealEndpointFromLiveOwner) { } ASSERT_FALSE(outcome->has_value()); EXPECT_EQ(outcome->error(), common::Error::kOperationFailed); - EXPECT_TRUE(thief_logger.Contains("served by another live process")); + // Either guard is a correct refusal, and which one fires is an implementation + // detail: EndpointLock runs first and will normally catch it, with the + // connect(2) probe behind it for an owner that holds no lock. + EXPECT_TRUE(thief_logger.Contains("Refusing to bind")); // The assertion that actually proves no hijack occurred: a fresh peer // connecting to the contested endpoint still reaches the original owner. @@ -287,4 +363,23 @@ TEST_F(ZmqTransportStartupTest, CreateSucceedsOverStaleSocketFile) { EXPECT_TRUE(outcome->value()->IsReady()); } +// Twelve processes contend for one bind endpoint, released together by a shared +// spin barrier so the decide-then-bind window is genuinely contended rather +// than spread out by process startup. +// +// Exactly one may win. Every additional winner is a process that believes it +// owns an endpoint libzmq has already unlinked out from under it -- they do not +// fail, which is precisely what makes this worth a test. +// +// Measured with EndpointLock disabled and only the connect(2) probe in place, +// this produces between 1 and 7 winners per run. That spread is the reason the +// lock exists: a probe followed by a bind is two syscalls with a window between +// them, and flock has no window at all. +TEST_F(ZmqTransportStartupTest, ConcurrentBindsProduceExactlyOneOwner) { + const auto contested = TrackForCleanup(UniqueEndpoint("contested")); + const int winners = CountConcurrentBindWinners(contested); + EXPECT_EQ(winners, 1) << winners << " processes each believe they own " + << contested; +} + } // namespace diff --git a/src/libs/mcu/host/zmq_transport.cpp b/src/libs/mcu/host/zmq_transport.cpp index 93531d2..286db9e 100644 --- a/src/libs/mcu/host/zmq_transport.cpp +++ b/src/libs/mcu/host/zmq_transport.cpp @@ -1,6 +1,9 @@ #include "zmq_transport.hpp" +#include +#include #include +#include #include #include @@ -21,6 +24,8 @@ namespace mcu { namespace { constexpr std::string_view kIpcScheme{"ipc://"}; +constexpr std::string_view kLockSuffix{".lock"}; +constexpr mode_t kLockFileMode{0600}; // RAII for a bare file descriptor. The liveness probe below is the only place // this file talks to POSIX sockets directly, and it must not leak an fd on any @@ -75,6 +80,40 @@ auto IpcPathHasLiveOwner(const std::string& path) -> bool { } // namespace +EndpointLock::~EndpointLock() { + if (fd_ >= 0) { + ::close(fd_); // Closing the fd is what releases the lock. + } +} + +auto EndpointLock::TryAcquire(const std::string& endpoint) -> bool { + const std::string_view endpoint_view{endpoint}; + if (!endpoint_view.starts_with(kIpcScheme)) { + return true; // No filesystem path to guard. + } + const std::string lock_path{ + std::string{endpoint_view.substr(kIpcScheme.size())} + + std::string{kLockSuffix}}; + + // The lock file is deliberately never unlinked. Removing it would reintroduce + // exactly the race it exists to close: one process unlinking the file another + // has already opened, leaving the two holding locks on different inodes and + // both believing they won. It stays behind as a zero-byte marker. + const int descriptor = + ::open(lock_path.c_str(), O_CREAT | O_RDWR | O_CLOEXEC, kLockFileMode); + if (descriptor < 0) { + // Cannot lock here -- a read-only directory, for instance. Fall through to + // the liveness probe rather than refusing to start over a missing luxury. + return true; + } + if (::flock(descriptor, LOCK_EX | LOCK_NB) != 0) { + ::close(descriptor); + return false; + } + fd_ = descriptor; + return true; +} + auto ZmqTransport::Create(const std::string& to_emulator, const std::string& from_emulator, Dispatcher& dispatcher, const TransportConfig& config) @@ -200,6 +239,10 @@ auto ZmqTransport::SetSocketOptions() -> void { // // libzmq gives us no way to ask it not to do that, so we check before handing // it the endpoint and refuse to start rather than become the thief. +// +// On its own this check is racy -- another process can bind between it and our +// bind. EndpointLock closes that window for anything using the same lock; this +// remains as the best available answer for an owner that is not. auto ZmqTransport::EndpointHasLiveOwner(const std::string& endpoint) const -> bool { const std::string_view endpoint_view{endpoint}; @@ -320,6 +363,16 @@ auto ZmqTransport::ServerThread(const std::string& endpoint) -> void { socket.set(zmq::sockopt::rcvtimeo, static_cast(config_.poll_timeout.count())); + // Two checks, and the order matters. The lock is the guarantee: it is + // atomic, so it excludes every other transport that plays by the same + // rules, with no window between deciding and binding. The probe is the + // fallback for an owner that does not -- an older build, or anything else + // that happens to be listening on that path. + if (!endpoint_lock_.TryAcquire(endpoint)) { + LogError("Refusing to bind: endpoint is locked by another live process"); + SignalBind(BindOutcome::kFailed); + return; + } if (EndpointHasLiveOwner(endpoint)) { LogError("Refusing to bind: endpoint is served by another live process"); SignalBind(BindOutcome::kFailed); diff --git a/src/libs/mcu/host/zmq_transport.hpp b/src/libs/mcu/host/zmq_transport.hpp index 1af43d2..0d989ac 100644 --- a/src/libs/mcu/host/zmq_transport.hpp +++ b/src/libs/mcu/host/zmq_transport.hpp @@ -78,6 +78,35 @@ struct TransportConfig { } }; +// Exclusive advisory ownership of a bind endpoint, held for the lifetime of the +// transport that took it. +// +// This is what makes "may I bind here" atomic. A connect(2) liveness probe +// cannot be: another process can bind in the window between the probe and our +// own bind, and libzmq will then unlink whichever socket file it finds. 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 SIGKILLed run leaves nothing behind +// that would block the next one. +class EndpointLock { + public: + EndpointLock() = default; + EndpointLock(const EndpointLock&) = delete; + EndpointLock(EndpointLock&&) = delete; + auto operator=(const EndpointLock&) -> EndpointLock& = delete; + auto operator=(EndpointLock&&) -> EndpointLock& = delete; + ~EndpointLock(); + + // Takes the lock guarding `endpoint`. False means another live process holds + // it. Endpoints with no lockable path succeed trivially. + auto TryAcquire(const std::string& endpoint) -> bool; + + private: + int fd_{-1}; +}; + class ZmqTransport : public Transport { public: ZmqTransport() = delete; @@ -153,6 +182,9 @@ class ZmqTransport : public Transport { zmq::context_t from_emulator_context_{1}; std::atomic running_{true}; + // Taken on the server thread, released when this object is destroyed -- after + // the destructor has joined that thread, so the two never race. + EndpointLock endpoint_lock_; BindOutcome bind_outcome_{BindOutcome::kPending}; // guarded by bind_mutex_ std::condition_variable bind_cv_; std::mutex bind_mutex_; From fc653731387570b31cdcd30b3f6497c526be6d45 Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 06:27:49 +0000 Subject: [PATCH 5/6] chore: ignore coverage counter files The Debug preset builds with CODE_COVERAGE, so every run of an instrumented binary drops a .profraw wherever it was invoked from -- usually the repo root, where it showed up as untracked noise after each test run. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.gitignore b/.gitignore index 8f91253..55bf81b 100644 --- a/.gitignore +++ b/.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/ From 53e0cde864bdbf72530cd7ded067a66cfe4d085b Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 06:37:50 +0000 Subject: [PATCH 6/6] fix(emulator): stop the emulator hijacking a live endpoint The Python side had the same defect as the C++ transport, and worse. libzmq unlinks an ipc path before binding it, so a second emulator would displace a running one and take its rendezvous name -- the original keeps its existing connections, because the inode outlives the name, but every later connect() reaches the newcomer, with no error on either side. On top of that, run() unlinked the socket file itself, unconditionally, which made the displacement certain rather than merely possible. That unlink is gone. It was never needed: libzmq's own unlink already makes a file left by a killed process harmless, which is the only case it was meant to cover. In its place, the same two guards the C++ side uses, in the same order. A new endpoint module carries an flock-based EndpointLock -- atomic, so nothing can slip between the check and the bind, and released by the kernel on process death, so a crashed run leaves nothing that blocks the next one -- with a connect(2) liveness probe behind it for an owner that holds no lock. The lock file naming deliberately matches zmq_transport.cpp; the two must agree or they do not exclude each other. start() now waits for the bind outcome rather than spinning on a flag until a timeout. A failure surfaces immediately with its cause instead of five seconds later as "failed to start within timeout", and run() publishes that outcome on every path out of the bind phase, so start() cannot wait for something that will never arrive. test_endpoint_ownership.py covers it: a second emulator refusing to start while the first keeps its endpoint (verified to fail against the previous unlink behaviour), a stale file not blocking startup, and a lock released by killing its holder. The last is the reason for flock over an O_EXCL marker, which would have needed its own liveness check -- the very problem being solved. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/host_emulator/emulator.py | 88 +++++++--- .../src/host_emulator/endpoint.py | 128 ++++++++++++++ .../tests/test_endpoint_ownership.py | 157 ++++++++++++++++++ 3 files changed, 350 insertions(+), 23 deletions(-) create mode 100644 py/host-emulator/src/host_emulator/endpoint.py create mode 100644 py/host-emulator/tests/test_endpoint_ownership.py diff --git a/py/host-emulator/src/host_emulator/emulator.py b/py/host-emulator/src/host_emulator/emulator.py index 67bef32..11051c2 100755 --- a/py/host-emulator/src/host_emulator/emulator.py +++ b/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,15 +209,24 @@ 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) diff --git a/py/host-emulator/src/host_emulator/endpoint.py b/py/host-emulator/src/host_emulator/endpoint.py new file mode 100644 index 0000000..16bacaa --- /dev/null +++ b/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 diff --git a/py/host-emulator/tests/test_endpoint_ownership.py b/py/host-emulator/tests/test_endpoint_ownership.py new file mode 100644 index 0000000..7ed5887 --- /dev/null +++ b/py/host-emulator/tests/test_endpoint_ownership.py @@ -0,0 +1,157 @@ +"""Endpoint ownership tests for the emulator. + +These cover the Python half of the same defect fixed in +``src/libs/mcu/host/zmq_transport.cpp``: libzmq unlinks an ipc path before +binding it, so a second emulator used to displace a running one silently. The +emulator additionally unlinked the path itself, unconditionally, which made it +certain rather than merely possible. + +These are pure emulator tests -- no application binary -- so they run without +any of the --blinky/--uart-echo/--i2c-demo options. +""" + +from __future__ import annotations + +import socket +import subprocess +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from host_emulator import DeviceEmulator +from host_emulator.endpoint import EndpointLock, endpoint_path, has_live_owner + +if TYPE_CHECKING: + from collections.abc import Generator + + +@pytest.fixture +def endpoints(tmp_path: Path) -> Generator[tuple[str, str]]: + """A unique endpoint pair, with its lock files cleaned up afterwards.""" + from_device = f"ipc://{tmp_path}/from_device.ipc" + to_device = f"ipc://{tmp_path}/to_device.ipc" + yield from_device, to_device + for endpoint in (from_device, to_device): + path = endpoint_path(endpoint) + if path is not None: + path.unlink(missing_ok=True) + path.with_name(path.name + ".lock").unlink(missing_ok=True) + + +def test_second_emulator_refuses_to_steal_endpoint( + endpoints: tuple[str, str], tmp_path: Path +) -> None: + """A second emulator must fail rather than displace a running one.""" + from_device, to_device = endpoints + first = DeviceEmulator(from_device, to_device) + first.start() + + try: + second = DeviceEmulator(from_device, f"ipc://{tmp_path}/to_device2.ipc") + try: + with pytest.raises(RuntimeError, match="another live process"): + second.start() + finally: + # Required, not tidiness: a DeviceEmulator opens its sockets in + # __init__, so one whose start() failed still holds them, and + # zmq_ctx_term blocks on an open socket. Skipping this hangs the + # interpreter at exit rather than failing the test. + second.stop() + + # The point of the test: the first emulator still owns the endpoint. + # Before the fix the second bound successfully and every subsequent + # connect() reached it instead, with no error on either side. + probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + probe.connect(str(endpoint_path(from_device))) + finally: + probe.close() + finally: + first.stop() + + +def test_stale_socket_file_does_not_block_startup( + endpoints: tuple[str, str], +) -> None: + """A path left by a killed process must not stop the next run. + + The liveness probe has to distinguish "someone is listening" from "a file + exists"; reading the latter as ownership would make any crash require + manual cleanup. + """ + from_device, to_device = endpoints + path = endpoint_path(from_device) + assert path is not None + + # Exactly what a SIGKILLed process leaves: bound, then closed without + # unlinking. + stale = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + stale.bind(str(path)) + stale.close() + assert path.exists() + + emulator = DeviceEmulator(from_device, to_device) + emulator.start() + try: + assert emulator.running + finally: + emulator.stop() + + +def test_lock_is_released_when_holder_dies(endpoints: tuple[str, str]) -> None: + """A killed holder must not leave the endpoint locked. + + This is why the lock is flock and not an O_EXCL lock file: the kernel drops + it when the fd closes, including on process death, so a crashed run needs no + cleanup. An O_EXCL marker would need its own liveness check to distinguish + held from abandoned -- the very problem the lock exists to solve. + """ + from_device, _ = endpoints + + # A real subprocess rather than multiprocessing: the default start method + # is forkserver, which pickles the target, and a locally-defined function + # cannot be pickled. + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import sys, time;" + "sys.path.insert(0, sys.argv[1]);" + "from host_emulator.endpoint import EndpointLock;" + "lock = EndpointLock();" + "print(lock.try_acquire(sys.argv[2]), flush=True);" + "time.sleep(60)", + str(Path(__file__).parent.parent / "src"), + from_device, + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "True" # holder has the lock + + # Contended while alive. + assert EndpointLock().try_acquire(from_device) is False + finally: + holder.kill() + holder.wait(timeout=5) + + # Free again the moment the holder is gone. + survivor = EndpointLock() + assert survivor.try_acquire(from_device) is True + survivor.release() + + +def test_has_live_owner_is_false_for_absent_endpoint(tmp_path: Path) -> None: + """Nothing there at all is not ownership.""" + assert has_live_owner(f"ipc://{tmp_path}/never_created.ipc") is False + + +def test_non_ipc_endpoints_are_not_guarded() -> None: + """tcp:// and inproc:// have no filesystem path, so both guards no-op.""" + assert endpoint_path("tcp://127.0.0.1:5555") is None + assert has_live_owner("tcp://127.0.0.1:5555") is False + assert EndpointLock().try_acquire("tcp://127.0.0.1:5555") is True