fix(rmw_wait): deliver TRANSIENT_LOCAL latched samples to late joiners of idle publishers

rmw_unix_socket_cpp/src/rmw_wait.cpp

@@ -395,6 +395,15 @@ rmw_ret_t rmw_wait(
       }
     }
 
+    // Bound the wait so an idle executor loops and re-runs the top-of-wait
+    // TRANSIENT_LOCAL late-joiner replay: a joining subscriber only bumps the
+    // shm generation counter (no fd fires), so an unbounded wait would never
+    // replay to it. Only shortens the timeout; a non-blocking 0 stays 0.
+    constexpr int TL_REPLAY_POLL_MS = 200;
+    if (timeout_ms < 0 || timeout_ms > TL_REPLAY_POLL_MS) {
+      timeout_ms = TL_REPLAY_POLL_MS;
+    }
+
     // Block, retrying on EINTR. A finite timeout uses a steady_clock deadline
     // so a signal interruption neither returns TIMEOUT early nor busy-loops.
     struct epoll_event ready_events[64];

rmw_unix_socket_cpp/test/test_rmw_qos.cpp

@@ -14,6 +14,7 @@
 
 #include "test_base.hpp"
 
+#include <atomic>
 #include <cstdlib>
 #include <cstring>
 #include <thread>
@@ -975,3 +976,92 @@ TEST_F(QosTest, MultipleClientsOneService)
   auto _r2 [[maybe_unused]] = rmw_destroy_client(node, cli1);
   auto _r3 [[maybe_unused]] = rmw_destroy_service(node, srv);
 }
+
+TEST_F(QosTest, TransientLocalReplayReachesLateJoinerWhileWaitBlocked)
+{
+  // The subscriber joins AFTER the publisher's executor is already blocked in
+  // rmw_wait. A joining subscriber only bumps the shm generation counter, which
+  // signals no fd, so an idle rmw_wait(infinite) would block in epoll forever
+  // and never re-run the top-of-wait replay. The latched message must still
+  // reach the late joiner within a bounded time.
+  auto qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE,
+    RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5);
+
+  // A service anchors ctx resolution inside rmw_wait, mirroring an idle node
+  // whose wait set holds only its services.
+  auto srv_ts = rosidl_typesupport_cpp::get_service_type_support_handle<
+    test_msgs::srv::BasicTypes>();
+  auto svc_qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_VOLATILE);
+  auto * srv = rmw_create_service(node, srv_ts, "/idle_anchor", &svc_qos);
+  ASSERT_NE(nullptr, srv);
+
+  // Guard condition only unblocks the executor thread on teardown so the test
+  // never hangs when the message never arrives (the failing case).
+  auto * gc = rmw_create_guard_condition(&context);
+  ASSERT_NE(nullptr, gc);
+
+  auto pub_opts = rmw_get_default_publisher_options();
+  auto * pub = rmw_create_publisher(node, ts, "/idle_replay", &qos, &pub_opts);
+  ASSERT_NE(nullptr, pub);
+
+  // Publish before any subscriber exists; then the node goes idle.
+  test_msgs::msg::BasicTypes m;
+  m.int32_value = 7;
+  EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr));
+
+  auto * ws = rmw_create_wait_set(&context, 4);
+  ASSERT_NE(nullptr, ws);
+
+  // Executor thread: spin rmw_wait with an INFINITE timeout, like an idle node.
+  std::atomic<bool> stop{false};
+  std::thread executor(
+    [&] {
+      while (!stop.load()) {
+        void * srv_array[1] = {srv->data};
+        rmw_services_t services;
+        services.services = srv_array;
+        services.service_count = 1;
+        void * gc_array[1] = {gc->data};
+        rmw_guard_conditions_t gcs;
+        gcs.guard_conditions = gc_array;
+        gcs.guard_condition_count = 1;
+        auto _r [[maybe_unused]] = rmw_wait(
+          nullptr, &gcs, &services, nullptr, nullptr, ws, nullptr);
+      }
+    });
+
+  // Let the executor reach epoll and block before the subscriber joins.
+  std::this_thread::sleep_for(std::chrono::milliseconds(150));
+
+  // Late joiner — created after the executor is already blocked in rmw_wait.
+  auto sub_opts = rmw_get_default_subscription_options();
+  auto * sub = rmw_create_subscription(node, ts, "/idle_replay", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  bool got = false;
+  for (int i = 0; i < 300 && !got; ++i) {
+    test_msgs::msg::BasicTypes recv;
+    bool taken = false;
+    if (rmw_take(sub, &recv, &taken, nullptr) == RMW_RET_OK && taken &&
+      recv.int32_value == 7)
+    {
+      got = true;
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+
+  stop.store(true);
+  auto _t [[maybe_unused]] = rmw_trigger_guard_condition(gc);
+  executor.join();
+
+  EXPECT_TRUE(got) << "late joiner never received the latched message while the "
+    "publisher's executor was blocked in rmw_wait";
+
+  auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
+  auto _r2 [[maybe_unused]] = rmw_destroy_wait_set(ws);
+  auto _r3 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
+  auto _r4 [[maybe_unused]] = rmw_destroy_guard_condition(gc);
+  auto _r5 [[maybe_unused]] = rmw_destroy_service(node, srv);
+}