feat: name the subscriber (node + topic) in the UDS send-drop warning and clarify the byte count

The drop warning printed only the peer's socket path and worded the
message size as if it were the receive-buffer fill level
("recv buffer full (N bytes)"). Both made a slow-subscriber drop hard
to diagnose: the path only encodes a PID, and the byte count looked
like it contradicted the configured buffer size.

Now the ENOBUFS/EAGAIN branch names the offending subscriber by its
fully-qualified node name and topic, states plainly that the number is
the dropped message's size, and shows the configured recv-buffer size
for reference.

The peer label is built once per graph change (in refresh_subscribers,
off the hot path) and stored in the copy-on-write subscriber cache
alongside the socket path, so the publish hot path still copies just a
refcount and never formats a string. send_to takes an optional
peer_label; the three publisher fan-out paths supply it,
request/response callers pass none and are unaffected.

Rewritten against current devel: the wake-side TRANSIENT_LOCAL replay
this originally also touched in rmw_wait.cpp no longer exists (#60's
pull-based replay), and the three cache-refresh sites it edited are now
the single refresh_subscribers helper.

Author: benaliabderrahmane

rmw_unix_socket_cpp/src/rmw_publisher.cpp

@@ -201,13 +201,14 @@ rmw_ret_t rmw_destroy_publisher(rmw_node_t * node, rmw_publisher_t * publisher)
   return RMW_RET_OK;
 }
 
-// Refresh the generation-keyed subscriber-path cache if `current_gen` moved,
-// and return the (possibly shared) path list. The memoization invariant that
-// keeps the pull-based replay proof sound: cached_generation only ever
+// Refresh the generation-keyed subscriber cache if `current_gen` moved, and
+// return the (possibly shared) subscriber list. The memoization invariant
+// that keeps the pull-based replay proof sound: cached_generation only ever
 // advances to a value loaded BEFORE the query ran, under sub_cache_mutex —
 // so current_gen == cached_generation implies the cache already saw every
 // subscriber whose registration that generation covers.
-static std::shared_ptr<const std::vector<std::string>> refresh_sub_paths(
+static std::shared_ptr<const std::vector<rmw_uds::CachedSubscriber>>
+refresh_subscribers(
   rmw_uds::UdsPublisher * pub_data,
   rmw_uds::RegistryHeader * header,
   uint64_t current_gen)
@@ -222,17 +223,21 @@ static std::shared_ptr<const std::vector<std::string>> refresh_sub_paths(
     auto subs = rmw_uds::registry_query(
       header, rmw_uds::ENTRY_SUBSCRIPTION, pub_data->topic_name.c_str(),
       nullptr, nullptr);
-    auto fresh = std::make_shared<std::vector<std::string>>();
+    auto fresh = std::make_shared<std::vector<rmw_uds::CachedSubscriber>>();
     fresh->reserve(subs.size());
     for (const auto & s : subs) {
       if (!s.socket_path.empty()) {
-        fresh->push_back(s.socket_path);
+        fresh->push_back(
+          {s.socket_path,
+            rmw_uds::make_peer_label(
+              s.node_namespace.c_str(), s.node_name.c_str(),
+              pub_data->topic_name.c_str())});
       }
     }
     pub_data->cached_generation = current_gen;
-    pub_data->cached_subscriber_paths = std::move(fresh);
+    pub_data->cached_subscribers = std::move(fresh);
   }
-  return pub_data->cached_subscriber_paths;
+  return pub_data->cached_subscribers;
 }
 
 // Latch a TRANSIENT_LOCAL sample into the pull cache and fan it out live.
@@ -296,7 +301,7 @@ static rmw_ret_t transient_local_publish(
 
     auto * header = rmw_uds::registry_header(pub_data->context->registry_ptr);
     const uint64_t current_gen = rmw_uds::registry_generation(header);
-    auto sub_paths = refresh_sub_paths(pub_data, header, current_gen);
+    auto targets = refresh_subscribers(pub_data, header, current_gen);
 
     // Live fan-out: payloads at or above the datagram threshold reuse the
     // durable segment the latch just committed (same bytes, same
@@ -311,11 +316,12 @@ static rmw_ret_t transient_local_publish(
       wire_size = sizeof(staged_desc);
     }
 
-    if (sub_paths) {
-      for (const auto & path : *sub_paths) {
+    if (targets) {
+      for (const auto & t : *targets) {
         if (rmw_uds::send_to(
             pub_data->context->send_socket_fd,
-            path, hdr, wire_data, wire_size) == rmw_uds::SendResult::ConfigError)
+            t.socket_path, hdr, wire_data, wire_size,
+            t.label.c_str()) == rmw_uds::SendResult::ConfigError)
         {
           config_error = true;
         }
@@ -380,7 +386,7 @@ rmw_ret_t rmw_publish(
   // since we last cached the subscriber list. The hot path is purely local.
   auto * header = rmw_uds::registry_header(pub_data->context->registry_ptr);
   uint64_t current_gen = rmw_uds::registry_generation(header);
-  auto sub_paths = refresh_sub_paths(pub_data, header, current_gen);
+  auto targets = refresh_subscribers(pub_data, header, current_gen);
 
   // Non-latched path: serialize once, choosing the destination by size — a
   // large payload is written directly into the reserved ring record (no heap
@@ -404,11 +410,12 @@ rmw_ret_t rmw_publish(
 
   // Surface EMSGSIZE; soft drops (EAGAIN/ENOENT) are logged in send_to.
   bool config_error = false;
-  if (sub_paths) {
-    for (const auto & path : *sub_paths) {
+  if (targets) {
+    for (const auto & t : *targets) {
       if (rmw_uds::send_to(
           pub_data->context->send_socket_fd,
-          path, hdr, wire.data, wire.size) == rmw_uds::SendResult::ConfigError)
+          t.socket_path, hdr, wire.data, wire.size,
+          t.label.c_str()) == rmw_uds::SendResult::ConfigError)
       {
         config_error = true;
       }
@@ -463,7 +470,7 @@ rmw_ret_t rmw_publish_serialized_message(
   // Reuse the cached subscriber list (refresh only on graph change)
   auto * header = rmw_uds::registry_header(pub_data->context->registry_ptr);
   uint64_t current_gen = rmw_uds::registry_generation(header);
-  auto sub_paths = refresh_sub_paths(pub_data, header, current_gen);
+  auto targets = refresh_subscribers(pub_data, header, current_gen);
 
   // Large non-TL payloads: stage once into the cycling ring, fan out descriptors.
   rmw_uds::ShmPayloadDescriptor desc;
@@ -472,11 +479,12 @@ rmw_ret_t rmw_publish_serialized_message(
     serialized_message->buffer, serialized_message->buffer_length, hdr, desc);
 
   bool config_error = false;
-  if (sub_paths) {
-    for (const auto & path : *sub_paths) {
+  if (targets) {
+    for (const auto & t : *targets) {
       if (rmw_uds::send_to(
           pub_data->context->send_socket_fd,
-          path, hdr, wire.data, wire.size) == rmw_uds::SendResult::ConfigError)
+          t.socket_path, hdr, wire.data, wire.size,
+          t.label.c_str()) == rmw_uds::SendResult::ConfigError)
       {
         config_error = true;
       }

rmw_unix_socket_cpp/src/transport.cpp

@@ -108,12 +108,33 @@ int create_send_socket()
   return fd;
 }
 
+std::string make_peer_label(
+  const char * node_namespace, const char * node_name, const char * topic)
+{
+  std::string out = "node '";
+  // ROS fully-qualified node name: namespace + '/' + name, with a "/" (root)
+  // namespace collapsing so we emit "/name" rather than "//name".
+  if (node_namespace && node_namespace[0] && std::strcmp(node_namespace, "/") != 0) {
+    out += node_namespace;
+  }
+  out += '/';
+  out += (node_name && node_name[0]) ? node_name : "<unknown>";
+  out += '\'';
+  if (topic && topic[0]) {
+    out += " on topic '";
+    out += topic;
+    out += '\'';
+  }
+  return out;
+}
+
 SendResult send_to(
   int send_fd,
   const std::string & dest_path,
   const WireHeader & header,
   const uint8_t * payload,
-  size_t payload_size)
+  size_t payload_size,
+  const char * peer_label)
 {
   struct sockaddr_un addr;
   std::memset(&addr, 0, sizeof(addr));
@@ -139,9 +160,9 @@ SendResult send_to(
   }
 
   // Classify the failure. Hot path — every category is throttled so a
-  // persistently-failing peer can't drown the log. The destination path
-  // encodes the peer's prefix+pid (see make_socket_path) which is enough
-  // to identify the offending subscriber from the message alone.
+  // persistently-failing peer can't drown the log. `peer_label` (when the
+  // caller supplies one) names the offending subscriber; the destination
+  // path also encodes its prefix+pid (see make_socket_path).
   const int err = errno;
   const size_t total = sizeof(WireHeader) + payload_size;
   SendResult result = SendResult::SoftDrop;
@@ -164,12 +185,15 @@ SendResult send_to(
     case ENOBUFS:
       // Peer's receive queue is full (or our send queue, briefly). This
       // is the classic "slow subscriber" symptom — the subscriber isn't
-      // calling rmw_take fast enough to drain its socket.
+      // calling rmw_take fast enough to drain its socket. The byte count is
+      // this dropped message's size, not the buffer's fill level.
       RMW_UDS_LOG_WARN_THROTTLE(
         1000,
-        "UDS send to '%s' dropped: subscriber recv buffer full (%zu bytes, errno=%s). "
+        "UDS send to %s (socket '%s') dropped: subscriber recv queue full — "
+        "dropped a %zu-byte message (recv buf configured %d MB, errno=%s). "
         "Slow subscriber or undersized SO_RCVBUF.",
-        dest_path.c_str(), total, std::strerror(err));
+        peer_label ? peer_label : "peer", dest_path.c_str(), total,
+        RECV_BUF_SIZE / (1024 * 1024), std::strerror(err));
       break;
     case ENOENT:
     case ECONNREFUSED:

rmw_unix_socket_cpp/src/transport.hpp

@@ -40,13 +40,23 @@ int create_send_socket();
 // ConfigError = EMSGSIZE (hard, surface to caller); SoftDrop = transient.
 enum class SendResult { Ok, SoftDrop, ConfigError };
 
-// Send a datagram (header + payload) to the given socket path.
+// Send a datagram (header + payload) to the given socket path. `peer_label`,
+// if non-null, is a human-readable peer identity (see make_peer_label) shown
+// in the drop/failure diagnostics; pass null when no identity is available.
 SendResult send_to(
   int send_fd,
   const std::string & dest_path,
   const WireHeader & header,
   const uint8_t * payload,
-  size_t payload_size);
+  size_t payload_size,
+  const char * peer_label = nullptr);
+
+// Build a human-readable peer identity for diagnostics, e.g.
+// "node '/ns/name' on topic '/topic'". Null/empty fields are omitted; a "/"
+// (root) namespace collapses to a single leading slash (ROS fully-qualified
+// name), so ("/", "name", ...) yields "/name" rather than "//name".
+std::string make_peer_label(
+  const char * node_namespace, const char * node_name, const char * topic);
 
 // Receive a single datagram from a non-blocking socket.
 // Returns true if a message was received, false if EAGAIN/EWOULDBLOCK or error.

rmw_unix_socket_cpp/src/types.hpp

@@ -207,6 +207,15 @@ struct UdsNode
   int32_t registry_index = -1;
 };
 
+// A matched subscriber cached on the publisher. `label` is a pre-formatted
+// human-readable identity (node + topic, see make_peer_label) used only in
+// diagnostic logs; built once per graph change so publish never formats it.
+struct CachedSubscriber
+{
+  std::string socket_path;
+  std::string label;
+};
+
 // Publisher data
 struct UdsPublisher
 {
@@ -221,14 +230,15 @@ struct UdsPublisher
   UdsContext * context = nullptr;
   UdsNode * node = nullptr;
 
-  // PERFORMANCE: cache the matching subscriber socket paths to avoid locking
-  // the registry on every publish. We re-query only when the registry's
-  // generation counter changes (graph topology actually changed).
+  // PERFORMANCE: cache the matching subscribers to avoid locking the registry
+  // on every publish. We re-query only when the registry's generation counter
+  // changes (graph topology actually changed).
   std::mutex sub_cache_mutex;
   uint64_t cached_generation = 0;
   // Copy-on-write: swapped wholesale on graph-generation change so the publish/
-  // wait hot path copies one refcount instead of N strings. Null until first refresh.
-  std::shared_ptr<const std::vector<std::string>> cached_subscriber_paths;
+  // wait hot path copies one refcount instead of N entries. Each entry carries
+  // its diagnostic label, built here (off the hot path). Null until first refresh.
+  std::shared_ptr<const std::vector<CachedSubscriber>> cached_subscribers;
 
   // TRANSIENT_LOCAL latched cache, pulled by late joiners themselves at
   // rmw_create_subscription (see shm_transport.hpp). cache_mutex serializes

rmw_unix_socket_cpp/test/test_transport.cpp

@@ -37,6 +37,21 @@ class TransportTest : public ::testing::Test
   }
 };
 
+TEST_F(TransportTest, MakePeerLabelFormatsNodeAndTopic)
+{
+  EXPECT_EQ(
+    "node '/ns/talker' on topic '/chatter'",
+    rmw_uds::make_peer_label("/ns", "talker", "/chatter"));
+  // Root namespace "/" collapses to a single leading slash (not "//talker").
+  EXPECT_EQ(
+    "node '/talker' on topic '/chatter'",
+    rmw_uds::make_peer_label("/", "talker", "/chatter"));
+  // Empty namespace behaves like root; a null topic drops the topic clause.
+  EXPECT_EQ("node '/talker'", rmw_uds::make_peer_label("", "talker", nullptr));
+  // All-null is tolerated (name falls back to a placeholder).
+  EXPECT_EQ("node '/<unknown>'", rmw_uds::make_peer_label(nullptr, nullptr, nullptr));
+}
+
 TEST_F(TransportTest, CreateBoundSocket)
 {
   auto path = rmw_uds::make_socket_path(domain_id, "test");