From a2b6759d83675d4e038fa339dc0018e5072083a5 Mon Sep 17 00:00:00 2001 From: Nehal Patel Date: Thu, 27 Aug 2026 13:30:33 +0000 Subject: [PATCH] fix(transport): make ZmqTransport::Send actually retry The retry loop was unreachable, for two independent reasons. cppzmq's send() reports an expired ZMQ_SNDTIMEO by returning an EMPTY result and throws error_t only for everything else. Send() looked for EAGAIN exclusively inside a catch block and treated the falsy return as a hard failure, so a send timeout came back as kOperationFailed without ever reaching the loop's deadline check. Classify the attempt instead, in a TrySendOnce helper that catches the timeout in both the forms ZMQ reports it. Independently, the timeouts left no room to retry: send_timeout bounds ONE attempt and retry.total_timeout bounds all of them, and both defaulted to 1000ms, so the first EAGAIN arrived at the deadline and max_attempts and retry_delay described behaviour that could not happen. Size the default per-attempt slice to fit (3 * 300ms + 2 * 10ms <= 1000ms), and enforce the invariant at construction by shrinking send_timeout for any caller-supplied config that violates it. Clamping beats rejecting: asking for three attempts within a second is coherent, and the arithmetic that makes it fit is ours to do rather than grounds for failing startup. Two tests, both driving real backpressure by filling the send queue past ZMQ_SNDHWM, since -- contrary to what the header used to claim -- a peerless PAIR socket does not go mute. connect() creates the outbound pipe whether or not the far end is reachable and libzmq queues into it, so sends to nobody succeed until the high-water mark. Corrected those comments too. Each test asserts the error, the retry count, and the elapsed time, so no single wrong answer satisfies all three; against the unfixed code both fail with kOperationFailed and zero retries. Co-Authored-By: Claude Opus 5 (1M context) --- src/libs/mcu/host/test_zmq_transport.cpp | 172 +++++++++++++++++++++++ src/libs/mcu/host/zmq_transport.cpp | 118 +++++++++++----- src/libs/mcu/host/zmq_transport.hpp | 36 ++++- 3 files changed, 285 insertions(+), 41 deletions(-) diff --git a/src/libs/mcu/host/test_zmq_transport.cpp b/src/libs/mcu/host/test_zmq_transport.cpp index cd818d9..f7c695b 100644 --- a/src/libs/mcu/host/test_zmq_transport.cpp +++ b/src/libs/mcu/host/test_zmq_transport.cpp @@ -1,17 +1,24 @@ #include #include +#include #include #include #include +#include +#include #include #include +#include #include +#include #include #include +#include #include #include "dispatcher.hpp" +#include "libs/common/logger.hpp" #include "zmq_transport.hpp" namespace mcu { @@ -126,5 +133,170 @@ TEST_F(ZmqTransportTest, SendReceive) { ASSERT_EQ(response.value(), "World"); } +// Counts log lines, so a test can assert how many times Send() went round its +// retry loop rather than merely that it eventually failed. The mutex guards +// against the server thread logging concurrently with the test thread. +class CountingLogger : 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 Count(std::string_view needle) const -> std::size_t { + const std::lock_guard lock(mutex_); + return static_cast( + std::ranges::count_if(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_; +}; + +// Drives Send() into genuine backpressure, and reports the send that met it. +// +// The obvious setup -- point the transport at an endpoint nobody binds -- does +// NOT work, despite the mute state PAIR sockets are documented to have. +// connect() creates the outbound pipe immediately whether or not a peer is +// reachable, and libzmq queues into it, so a peerless send returns success. +// Backpressure begins only once ZMQ_SNDHWM messages (1000 by default) are +// outstanding, and that is where the retry loop finally has something to +// absorb. Measuring the first send to block gives the retry path a clean +// reading: every send before it succeeded on its first attempt and logged +// nothing. +struct BlockedSend { + std::expected result; + std::chrono::steady_clock::duration elapsed; + int sends; +}; + +auto SendUntilQueueBlocks(ZmqTransport& transport) -> BlockedSend { + // Comfortably past the default high-water mark, and a backstop against a + // regression that made sends succeed forever rather than hanging the suite. + constexpr int kMaxSends{5000}; + + BlockedSend blocked{}; + for (blocked.sends = 1; blocked.sends <= kMaxSends; ++blocked.sends) { + const auto start{std::chrono::steady_clock::now()}; + blocked.result = transport.Send("Hello"); + blocked.elapsed = std::chrono::steady_clock::now() - start; + if (!blocked.result) { + break; + } + } + return blocked; +} + +class ZmqTransportRetryTest : public ::testing::Test { + protected: + void TearDown() override { + std::error_code error{}; + const std::string path{ + own_endpoint_.substr(std::string_view{"ipc://"}.size())}; + std::filesystem::remove(path, error); + std::filesystem::remove(path + ".lock", error); + } + + static auto MakeConfig( + common::Logger& logger, std::chrono::milliseconds send_timeout, + std::chrono::milliseconds total_timeout) -> TransportConfig { + // TransportConfig has user-provided constructors, so it is not an + // aggregate: designated initialisers will not compile. Assign after + // construction. + TransportConfig config{logger}; + config.send_timeout = send_timeout; + config.retry.max_attempts = kAttempts; + config.retry.retry_delay = kRetryDelay; + config.retry.total_timeout = total_timeout; + return config; + } + + static constexpr uint32_t kAttempts{3}; + static constexpr std::chrono::milliseconds kRetryDelay{10}; + + const std::string absent_peer_endpoint_{Endpoint("absent_peer")}; + const std::string own_endpoint_{Endpoint("retry_own")}; + CountingLogger logger_; + const ReceiverMap receiver_map_; +}; + +// The regression test the fix owes: a blocked send must make every attempt it +// was configured for. +// +// Three assertions that fail independently, because each admits a different +// wrong answer on its own. kTimeout says the outcome was classified as +// retryable at all -- before the fix an expired ZMQ_SNDTIMEO came back as a +// falsy result rather than an exception and was reported as kOperationFailed, +// so the loop never even reached its deadline check. The log count says the +// loop iterated. The elapsed time says each iteration blocked on the socket for +// its own timeout instead of failing instantly. +TEST_F(ZmqTransportRetryTest, SendAttemptsEveryRetryWhenSocketBlocks) { + constexpr std::chrono::milliseconds kSendTimeout{100}; + constexpr std::chrono::milliseconds kTotalTimeout{2000}; + + Dispatcher dispatcher{receiver_map_}; + auto config = MakeConfig(logger_, kSendTimeout, kTotalTimeout); + auto transport = ZmqTransport::Create(absent_peer_endpoint_, own_endpoint_, + dispatcher, config); + ASSERT_TRUE(transport.has_value()); + ASSERT_TRUE((*transport)->IsReady()); + + const auto blocked = SendUntilQueueBlocks(**transport); + + ASSERT_FALSE(blocked.result.has_value()) + << "no send ever met backpressure in " << blocked.sends << " attempts"; + EXPECT_EQ(blocked.result.error(), common::Error::kTimeout); + EXPECT_EQ(logger_.Count("Send retrying"), kAttempts - 1) + << "the retry loop ran " << logger_.Count("Send retrying") + 1 + << " of its " << kAttempts << " configured attempts"; + // Lower bound only: an upper bound here would measure the machine's load + // rather than this code. Two attempts' worth of blocking cannot fit in one. + EXPECT_GE(blocked.elapsed, 2 * kSendTimeout); + EXPECT_LT(blocked.elapsed, kTotalTimeout) << "the whole-send budget overran"; +} + +// The shape of the configuration that shipped: one attempt allowed to consume +// the entire retry budget, so the first EAGAIN arrives at the deadline and +// max_attempts and retry_delay describe nothing that can happen. +// +// The transport now shrinks the per-attempt slice until the configured attempts +// fit, so the caller's stated intent -- three tries within this budget -- is +// what they get. +TEST_F(ZmqTransportRetryTest, SendRetriesWhenSendTimeoutClaimsTheWholeBudget) { + constexpr std::chrono::milliseconds kBudget{600}; + // (600ms - 2 * 10ms of retry delay) / 3 attempts. + constexpr std::chrono::milliseconds kExpectedSlice{193}; + + Dispatcher dispatcher{receiver_map_}; + auto config = MakeConfig(logger_, kBudget, kBudget); + auto transport = ZmqTransport::Create(absent_peer_endpoint_, own_endpoint_, + dispatcher, config); + ASSERT_TRUE(transport.has_value()); + ASSERT_TRUE((*transport)->IsReady()); + + const auto blocked = SendUntilQueueBlocks(**transport); + + ASSERT_FALSE(blocked.result.has_value()) + << "no send ever met backpressure in " << blocked.sends << " attempts"; + EXPECT_EQ(blocked.result.error(), common::Error::kTimeout); + EXPECT_EQ(logger_.Count("send_timeout exceeds"), 1U) + << "an incoherent budget should be clamped, and said so"; + EXPECT_EQ(logger_.Count("Send retrying"), kAttempts - 1); + // Bounded both ways, each side ruling out a different wrong answer. Below: + // the attempts genuinely blocked rather than failing instantly. Above: they + // were shortened to share the budget, not repeated at full length -- which + // would have taken three times as long. + EXPECT_GE(blocked.elapsed, 2 * kExpectedSlice); + EXPECT_LT(blocked.elapsed, 2 * kBudget); +} + } // namespace } // namespace mcu diff --git a/src/libs/mcu/host/zmq_transport.cpp b/src/libs/mcu/host/zmq_transport.cpp index 286db9e..708ecbb 100644 --- a/src/libs/mcu/host/zmq_transport.cpp +++ b/src/libs/mcu/host/zmq_transport.cpp @@ -7,7 +7,9 @@ #include #include +#include #include +#include #include #include #include @@ -151,6 +153,7 @@ ZmqTransport::ZmqTransport(const std::string& to_emulator, // NOLINT // an exception leaving this constructor destroys a *joinable* std::thread, // which calls std::terminate -- Create()'s handler never gets a look in. try { + ClampSendTimeoutToRetryBudget(); SetSocketOptions(); // Start server thread FIRST (it will BIND) @@ -212,6 +215,38 @@ auto ZmqTransport::SignalBind(BindOutcome outcome) -> void { bind_cv_.notify_all(); } +// Makes RetryConfig's invariant true rather than merely documented. +// +// ZMQ_SNDTIMEO bounds a single attempt; retry.total_timeout bounds the whole +// Send(). If one attempt may consume the entire budget then the first EAGAIN +// arrives at or after the deadline and Send() returns having tried once -- +// max_attempts and retry_delay become dead configuration. That was the shipped +// default: send_timeout and total_timeout were both 1000ms. +// +// Shrinking the per-attempt slice is preferred to rejecting the config. A +// caller who asks for three attempts within a second has said something +// coherent about what they want; the arithmetic that makes it fit is ours to +// do, and failing startup over a tuning number would be a worse answer. +auto ZmqTransport::ClampSendTimeoutToRetryBudget() -> void { + // Zero attempts would skip the loop entirely and report a timeout without + // ever touching the socket. + config_.retry.max_attempts = std::max(config_.retry.max_attempts, 1U); + const auto attempts{config_.retry.max_attempts}; + + const auto delays{(attempts - 1) * config_.retry.retry_delay}; + const auto sending{delays < config_.retry.total_timeout + ? config_.retry.total_timeout - delays + : std::chrono::milliseconds::zero()}; + // A floor of 1ms, because 0 means "never block" and a negative value means + // "block forever" -- both worse than a very short attempt. + const auto slice{std::max(std::chrono::milliseconds{1}, sending / attempts)}; + + if (config_.send_timeout > slice) { + LogWarning("send_timeout exceeds the per-attempt retry budget; clamping"); + config_.send_timeout = slice; + } +} + auto ZmqTransport::SetSocketOptions() -> void { // Set linger to 0 to discard messages immediately on close to_emulator_socket_.set(zmq::sockopt::linger, config_.linger_ms); @@ -292,6 +327,30 @@ ZmqTransport::~ZmqTransport() { } } +// One attempt at the socket, classified. +// +// The classification is the substance here. cppzmq's send() reports an expired +// ZMQ_SNDTIMEO by returning an EMPTY result and throws error_t only for +// everything else -- so the timeout this retry loop exists to absorb is a falsy +// return, not an exception. The previous code looked for it exclusively in a +// catch block, treated the falsy return as a hard failure, and so left the +// retry path unreachable even once the timeouts allowed for it. +// +// ETIMEDOUT is still caught for the same outcome: no libzmq version in use +// raises it here, but it means precisely what EAGAIN means and costs one line. +auto ZmqTransport::TrySendOnce(std::string_view data) -> SendAttempt { + try { + const auto result{ + to_emulator_socket_.send(zmq::buffer(data), zmq::send_flags::none)}; + return result ? SendAttempt::kSent : SendAttempt::kWouldBlock; + } catch (const zmq::error_t& e) { + if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { + return SendAttempt::kWouldBlock; + } + return SendAttempt::kFailed; + } +} + auto ZmqTransport::Send(std::string_view data) -> std::expected { if (state_.load() != TransportState::kReady) { @@ -299,53 +358,40 @@ auto ZmqTransport::Send(std::string_view data) return std::unexpected(common::Error::kInvalidState); } - // Calculate deadline for retry timeout + // The whole-send budget. Each attempt is separately bounded by the socket's + // ZMQ_SNDTIMEO, which the constructor sized to fit max_attempts of them in + // here; this deadline is the backstop for a peer that keeps us just under it. const auto deadline{std::chrono::steady_clock::now() + config_.retry.total_timeout}; - // Retry loop for (uint32_t attempt = 0; attempt < config_.retry.max_attempts; ++attempt) { - try { - auto result{ - to_emulator_socket_.send(zmq::buffer(data), zmq::send_flags::none)}; - if (result) { + switch (TrySendOnce(data)) { + case SendAttempt::kSent: if (attempt > 0) { LogDebug("Send succeeded after retry"); } - return {}; // Success! - } - } catch (const zmq::error_t& e) { - // Check if error is retryable - if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { - // Check if we've exceeded total timeout - if (std::chrono::steady_clock::now() >= deadline) { - LogError("Send timeout after retries"); - return std::unexpected(common::Error::kTimeout); - } - - if (attempt + 1 < config_.retry.max_attempts) { - LogDebug("Send retrying after transient error"); - } - - // Wait before retry (unless this was the last attempt) - if (attempt + 1 < config_.retry.max_attempts) { - std::this_thread::sleep_for(config_.retry.retry_delay); - } - continue; // Retry - } - - // Non-retryable error - LogError("Send failed with non-retryable error"); - return std::unexpected(common::Error::kOperationFailed); + return {}; + case SendAttempt::kFailed: + LogError("Send failed with non-retryable error"); + return std::unexpected(common::Error::kOperationFailed); + case SendAttempt::kWouldBlock: + break; } - // result was false but no exception - operation failed - LogError("Send operation returned false"); - return std::unexpected(common::Error::kOperationFailed); + if (attempt + 1 >= config_.retry.max_attempts) { + break; // Out of attempts. + } + if (std::chrono::steady_clock::now() >= deadline) { + break; // Out of budget. + } + LogDebug("Send retrying after transient error"); + std::this_thread::sleep_for(config_.retry.retry_delay); } - // Max attempts exceeded - LogError("Send failed: max attempts exceeded"); + // Every exhausted path is a timeout: a mute PAIR socket blocks rather than + // dropping, so running out of attempts and running out of budget both mean + // the peer never took the message. + LogError("Send timeout after retries"); return std::unexpected(common::Error::kTimeout); } diff --git a/src/libs/mcu/host/zmq_transport.hpp b/src/libs/mcu/host/zmq_transport.hpp index 0d989ac..852a813 100644 --- a/src/libs/mcu/host/zmq_transport.hpp +++ b/src/libs/mcu/host/zmq_transport.hpp @@ -25,8 +25,8 @@ namespace mcu { // 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. +// which is why a missing emulator surfaces as a failed Send() or Receive() +// 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. @@ -37,14 +37,30 @@ enum class TransportState : uint8_t { // 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. + // returns before any peer exists. + // + // Nor does a successful Send() in this state mean one is: connect() creates + // the outbound pipe whether or not the far end is reachable, and libzmq + // queues into it, so sends to nobody succeed until ZMQ_SNDHWM messages are + // outstanding. Only past that point does a send block 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, }; +// Bounds Send()'s retry loop. Read together with TransportConfig::send_timeout, +// which is the budget for ONE attempt while total_timeout is the budget for all +// of them: the two are only meaningful if +// +// max_attempts * send_timeout + (max_attempts - 1) * retry_delay +// <= total_timeout +// +// Give a single attempt the whole budget and the first EAGAIN already arrives +// at the deadline, so Send() gives up having tried exactly once and the two +// fields below describe behaviour that cannot happen. ZmqTransport enforces the +// invariant at construction by shrinking send_timeout to fit. struct RetryConfig { uint32_t max_attempts{3}; std::chrono::milliseconds retry_delay{10}; @@ -58,7 +74,10 @@ struct TransportConfig { // 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}; + // One attempt's worth of ZMQ_SNDTIMEO, not the whole send. The default is + // sized so all of retry.max_attempts fit inside retry.total_timeout with the + // inter-attempt delays: 3 * 300ms + 2 * 10ms = 920ms <= 1000ms. + std::chrono::milliseconds send_timeout{300}; std::chrono::milliseconds recv_timeout{5000}; int linger_ms{0}; // Discard pending messages on close RetryConfig retry{}; @@ -152,6 +171,13 @@ class ZmqTransport : public Transport { private: enum class BindOutcome : uint8_t { kPending, kBound, kFailed }; + // How one send attempt ended. kWouldBlock is the only retryable outcome, and + // it deliberately covers both ways ZMQ reports a full/mute socket -- see + // TrySendOnce. + enum class SendAttempt : uint8_t { kSent, kWouldBlock, kFailed }; + + auto TrySendOnce(std::string_view data) -> SendAttempt; + auto ClampSendTimeoutToRetryBudget() -> void; auto ServerThread(const std::string& endpoint) -> void; auto ServeLoop(zmq::socket_t& socket) -> void; auto SignalBind(BindOutcome outcome) -> void;