fix(transport): make ZmqTransport::Send actually retry

src/libs/mcu/host/test_zmq_transport.cpp

@@ -1,17 +1,24 @@
 #include <gtest/gtest.h>
 #include <unistd.h>
 
+#include <algorithm>
 #include <array>
 #include <atomic>
 #include <chrono>
+#include <cstddef>
+#include <expected>
 #include <filesystem>
 #include <libs/common/error.hpp>
+#include <mutex>
 #include <string>
+#include <string_view>
 #include <system_error>
 #include <thread>
+#include <vector>
 #include <zmq.hpp>
 
 #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<std::mutex> lock(mutex_);
+    return static_cast<std::size_t>(
+        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<std::mutex> lock(mutex_);
+    messages_.emplace_back(msg);
+  }
+
+  mutable std::mutex mutex_;
+  std::vector<std::string> 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<void, common::Error> 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

src/libs/mcu/host/zmq_transport.cpp

@@ -7,7 +7,9 @@
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <algorithm>
 #include <cerrno>
+#include <chrono>
 #include <expected>
 #include <filesystem>
 #include <string>
@@ -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,60 +327,71 @@ 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<void, common::Error> {
   if (state_.load() != TransportState::kReady) {
     LogWarning("Send failed: transport not ready");
     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);
 }