fix: replay TRANSIENT_LOCAL cache to late-joining subscribers

The publisher previously only replayed its cached messages inside
rmw_publish, so a subscriber that joined after the publisher had gone
idle never received the cache. In production this manifested as
late-joining subscribers (e.g. `ros2 topic echo`, bridge consumers)
silently missing latched topics such as camera_list.

Track TRANSIENT_LOCAL publishers on the context and drive cache replay
from rmw_wait whenever the registry generation changes. The mutex is
held across the loop so rmw_destroy_publisher cannot free a publisher
mid-iteration. known_subscriber_paths still deduplicates: whichever
path (publish or wait) runs first marks the path, the other no-ops.

Tests:
- TransientLocalReplayOnWaitNoSubsequentPublish: publisher publishes
  N messages then goes idle; subscriber joins, calls rmw_wait once,
  must receive all N cached messages without a fresh publish.
- TransientLocalReplayOnWaitDepthOne: same flow with depth=1.

Both tests fail on HEAD and pass with this change.

Author: benaliabderrahmane

rmw_unix_socket_cpp/src/rmw_publisher.cpp

@@ -5,6 +5,7 @@
 #include "transport.hpp"
 #include "types.hpp"
 
+#include <algorithm>
 #include <chrono>
 #include <cstring>
 
@@ -129,6 +130,12 @@ rmw_publisher_t * rmw_create_publisher(
   pub->topic_name = rcutils_strdup(topic_name, node->context->options.allocator);
   pub->options = publisher_options ? *publisher_options : rmw_get_default_publisher_options();
   pub->can_loan_messages = false;
+
+  if (pub_data->qos.durability == RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL) {
+    std::lock_guard<std::mutex> lock(ctx->transient_local_pubs_mutex);
+    ctx->transient_local_pubs.push_back(pub_data);
+  }
+
   return pub;
 }
 
@@ -142,6 +149,14 @@ rmw_ret_t rmw_destroy_publisher(rmw_node_t * node, rmw_publisher_t * publisher)
 
   auto * pub_data = static_cast<rmw_uds::UdsPublisher *>(publisher->data);
   if (pub_data) {
+    if (pub_data->context &&
+      pub_data->qos.durability == RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL)
+    {
+      auto * ctx = pub_data->context;
+      std::lock_guard<std::mutex> lock(ctx->transient_local_pubs_mutex);
+      auto & v = ctx->transient_local_pubs;
+      v.erase(std::remove(v.begin(), v.end(), pub_data), v.end());
+    }
     if (pub_data->context && pub_data->registry_index >= 0) {
       auto * header = rmw_uds::registry_header(pub_data->context->registry_ptr);
       rmw_uds::registry_remove(header, pub_data->registry_index);

rmw_unix_socket_cpp/src/rmw_wait.cpp

@@ -170,6 +170,44 @@ rmw_ret_t rmw_wait(
     uint64_t gen = rmw_uds::registry_generation(header);
     if (gen != ctx->last_registry_generation) {
       ctx->last_registry_generation = gen;
+
+      // TRANSIENT_LOCAL late-joiner replay. Lock held across the loop —
+      // rmw_destroy_publisher takes the same mutex then deletes, so this
+      // is what keeps each pub pointer alive while we dereference it.
+      std::lock_guard<std::mutex> tl_lock(ctx->transient_local_pubs_mutex);
+      for (auto * pub : ctx->transient_local_pubs) {
+        std::vector<std::string> sub_paths;
+        {
+          std::lock_guard<std::mutex> sc_lock(pub->sub_cache_mutex);
+          if (rmw_uds::registry_generation(header) != pub->cached_generation) {
+            auto subs = rmw_uds::registry_query(
+              header, rmw_uds::ENTRY_SUBSCRIPTION,
+              pub->topic_name.c_str(), nullptr, nullptr);
+            pub->cached_generation = rmw_uds::registry_generation(header);
+            pub->cached_subscriber_paths.clear();
+            pub->cached_subscriber_paths.reserve(subs.size());
+            for (const auto & s : subs) {
+              if (!s.socket_path.empty()) {
+                pub->cached_subscriber_paths.push_back(s.socket_path);
+              }
+            }
+          }
+          sub_paths = pub->cached_subscriber_paths;
+        }
+        std::lock_guard<std::mutex> c_lock(pub->cache_mutex);
+        for (const auto & path : sub_paths) {
+          if (pub->known_subscriber_paths.count(path) != 0) {
+            continue;
+          }
+          pub->known_subscriber_paths.insert(path);
+          for (const auto & cm : pub->message_cache) {
+            rmw_uds::send_to(
+              ctx->send_socket_fd,
+              path, cm.header, cm.payload.data(), cm.payload.size());
+          }
+        }
+      }
+
       // Trigger all graph guard conditions in the guard_conditions list
       if (guard_conditions) {
         for (size_t i = 0; i < guard_conditions->guard_condition_count; ++i) {

rmw_unix_socket_cpp/src/types.hpp

@@ -75,6 +75,9 @@ struct ReceivedMessage
   int64_t received_timestamp_ns;
 };
 
+// Forward declaration: publisher type defined below.
+struct UdsPublisher;
+
 // Per-context implementation data
 struct UdsContext
 {
@@ -86,6 +89,10 @@ struct UdsContext
   std::atomic<bool> is_shutdown{false};
   uint64_t last_registry_generation = 0;
   rmw_guard_condition_t * graph_guard_condition = nullptr;
+
+  // TRANSIENT_LOCAL publishers, for wait-side cache replay on graph change.
+  std::mutex transient_local_pubs_mutex;
+  std::vector<UdsPublisher *> transient_local_pubs;
 };
 
 // Node data

rmw_unix_socket_cpp/test/test_rmw_qos.cpp

@@ -182,6 +182,62 @@ TEST_F(QosTest, PublishReturnsErrorOnEMSGSIZE)
   auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
 }
 
+TEST_F(QosTest, TransientLocalReplayOnWaitNoSubsequentPublish)
+{
+  // Late sub must receive cached msgs from rmw_wait alone — no fresh publish.
+  auto qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE,
+    RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5);
+
+  auto pub_opts = rmw_get_default_publisher_options();
+  auto * pub = rmw_create_publisher(node, ts, "/wait_replay", &qos, &pub_opts);
+  ASSERT_NE(nullptr, pub);
+
+  for (int i = 1; i <= 3; ++i) {
+    test_msgs::msg::BasicTypes m;
+    m.int32_value = i;
+    EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr));
+  }
+
+  auto sub_opts = rmw_get_default_subscription_options();
+  auto * sub = rmw_create_subscription(node, ts, "/wait_replay", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  auto * ws = rmw_create_wait_set(&context, 1);
+  ASSERT_NE(nullptr, ws);
+
+  rmw_subscriptions_t subscriptions;
+  void * sub_array[1] = {sub->data};
+  subscriptions.subscribers = sub_array;
+  subscriptions.subscriber_count = 1;
+
+  rmw_time_t timeout;
+  timeout.sec = 0;
+  timeout.nsec = 500 * 1000 * 1000;  // 500 ms
+
+  EXPECT_EQ(RMW_RET_OK, rmw_wait(
+      &subscriptions, nullptr, nullptr, nullptr, nullptr, ws, &timeout));
+  EXPECT_NE(nullptr, subscriptions.subscribers[0]);
+
+  std::vector<int> received;
+  for (int i = 0; i < 10; ++i) {
+    test_msgs::msg::BasicTypes recv;
+    bool taken = false;
+    EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv, &taken, nullptr));
+    if (!taken) {break;}
+    received.push_back(recv.int32_value);
+  }
+
+  ASSERT_EQ(3u, received.size());
+  EXPECT_EQ(1, received[0]);
+  EXPECT_EQ(2, received[1]);
+  EXPECT_EQ(3, received[2]);
+
+  EXPECT_EQ(RMW_RET_OK, rmw_destroy_wait_set(ws));
+  auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
+  auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
+}
+
 TEST_F(QosTest, PublishStillReturnsOkOnSoftDropPeerGone)
 {
   // ENOENT on a vanished peer must stay RET_OK — only EMSGSIZE escalates.
@@ -212,6 +268,51 @@ TEST_F(QosTest, PublishStillReturnsOkOnSoftDropPeerGone)
   auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
 }
 
+TEST_F(QosTest, TransientLocalReplayOnWaitDepthOne)
+{
+  // depth=1 + late-join: the single cached msg must reach the new sub via wait.
+  auto qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE,
+    RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 1);
+
+  auto pub_opts = rmw_get_default_publisher_options();
+  auto * pub = rmw_create_publisher(node, ts, "/depth1_wait", &qos, &pub_opts);
+  ASSERT_NE(nullptr, pub);
+
+  test_msgs::msg::BasicTypes m;
+  m.int32_value = 42;
+  EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr));
+
+  auto sub_opts = rmw_get_default_subscription_options();
+  auto * sub = rmw_create_subscription(node, ts, "/depth1_wait", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  auto * ws = rmw_create_wait_set(&context, 1);
+  ASSERT_NE(nullptr, ws);
+
+  rmw_subscriptions_t subscriptions;
+  void * sub_array[1] = {sub->data};
+  subscriptions.subscribers = sub_array;
+  subscriptions.subscriber_count = 1;
+
+  rmw_time_t timeout;
+  timeout.sec = 0;
+  timeout.nsec = 500 * 1000 * 1000;
+  EXPECT_EQ(RMW_RET_OK, rmw_wait(
+      &subscriptions, nullptr, nullptr, nullptr, nullptr, ws, &timeout));
+  EXPECT_NE(nullptr, subscriptions.subscribers[0]);
+
+  test_msgs::msg::BasicTypes recv;
+  bool taken = false;
+  EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv, &taken, nullptr));
+  ASSERT_TRUE(taken);
+  EXPECT_EQ(42, recv.int32_value);
+
+  EXPECT_EQ(RMW_RET_OK, rmw_destroy_wait_set(ws));
+  auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
+  auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
+}
+
 TEST_F(QosTest, VolatileNoCache)
 {
   auto qos = make_qos(