fix: surface EMSGSIZE as RMW_RET_ERROR on publish

rmw_unix_socket_cpp/src/rmw_init.cpp

@@ -108,6 +108,8 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
     rmw_uds::cleanup_orphan_socket_files(domain_id);
   }
 
+  rmw_uds::warn_if_sysctl_buffers_undersized();
+
   // Create send socket
   ctx->send_socket_fd = rmw_uds::create_send_socket();
   if (ctx->send_socket_fd < 0) {

rmw_unix_socket_cpp/src/rmw_publisher.cpp

@@ -243,11 +243,21 @@ rmw_ret_t rmw_publish(
     }
   }
 
-  // Send current message to each subscriber (no registry lock held)
+  // Surface EMSGSIZE; soft drops (EAGAIN/ENOENT) are logged in send_to.
+  bool config_error = false;
   for (const auto & path : sub_paths) {
-    rmw_uds::send_to(
-      pub_data->context->send_socket_fd,
-      path, hdr, payload.data(), payload.size());
+    if (rmw_uds::send_to(
+        pub_data->context->send_socket_fd,
+        path, hdr, payload.data(), payload.size()) == rmw_uds::SendResult::ConfigError)
+    {
+      config_error = true;
+    }
+  }
+  if (config_error) {
+    RMW_SET_ERROR_MSG(
+      "UDS send: message too large for kernel send buffer (EMSGSIZE). "
+      "Raise net.core.wmem_max or reduce message size.");
+    return RMW_RET_ERROR;
   }
 
   return RMW_RET_OK;
@@ -297,10 +307,21 @@ rmw_ret_t rmw_publish_serialized_message(
     sub_paths = pub_data->cached_subscriber_paths;
   }
 
+  bool config_error = false;
   for (const auto & path : sub_paths) {
-    rmw_uds::send_to(
-      pub_data->context->send_socket_fd,
-      path, hdr, serialized_message->buffer, serialized_message->buffer_length);
+    if (rmw_uds::send_to(
+        pub_data->context->send_socket_fd,
+        path, hdr, serialized_message->buffer,
+        serialized_message->buffer_length) == rmw_uds::SendResult::ConfigError)
+    {
+      config_error = true;
+    }
+  }
+  if (config_error) {
+    RMW_SET_ERROR_MSG(
+      "UDS send: serialized message too large for kernel send buffer (EMSGSIZE). "
+      "Raise net.core.wmem_max or reduce message size.");
+    return RMW_RET_ERROR;
   }
 
   return RMW_RET_OK;

rmw_unix_socket_cpp/src/transport.cpp

@@ -5,6 +5,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
+#include <mutex>
 
 #include <dirent.h>
 #include <sys/socket.h>
@@ -91,7 +92,7 @@ int create_send_socket()
   return fd;
 }
 
-bool send_to(
+SendResult send_to(
   int send_fd,
   const std::string & dest_path,
   const WireHeader & header,
@@ -118,7 +119,7 @@ bool send_to(
 
   ssize_t sent = sendmsg(send_fd, &msg, MSG_DONTWAIT | MSG_NOSIGNAL);
   if (sent >= 0) {
-    return true;
+    return SendResult::Ok;
   }
 
   // Classify the failure. Hot path — every category is throttled so a
@@ -127,6 +128,7 @@ bool send_to(
   // to identify the offending subscriber from the message alone.
   const int err = errno;
   const size_t total = sizeof(WireHeader) + payload_size;
+  SendResult result = SendResult::SoftDrop;
   switch (err) {
     case EMSGSIZE:
       // Message exceeds the kernel's per-datagram cap (typically the send
@@ -137,6 +139,7 @@ bool send_to(
         "UDS send to '%s' failed: message too big (%zu bytes incl. header). "
         "Raise net.core.{wmem_max,rmem_max} or split the message.",
         dest_path.c_str(), total);
+      result = SendResult::ConfigError;
       break;
     case EAGAIN:
 #if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN
@@ -169,7 +172,7 @@ bool send_to(
         dest_path.c_str(), std::strerror(err), err, total);
       break;
   }
-  return false;
+  return result;
 }
 
 bool recv_from(
@@ -318,4 +321,41 @@ void cleanup_orphan_socket_files(size_t domain_id)
   closedir(dir);
 }
 
+void warn_if_sysctl_buffers_undersized()
+{
+  static std::once_flag once;
+  std::call_once(once, []() {
+      auto read_proc_int = [](const char * path) -> long {
+          FILE * f = std::fopen(path, "r");
+          if (!f) {
+            return -1;
+          }
+          long v = -1;
+          if (std::fscanf(f, "%ld", &v) != 1) {
+            v = -1;
+          }
+          std::fclose(f);
+          return v;
+        };
+
+      const long wmax = read_proc_int("/proc/sys/net/core/wmem_max");
+      const long rmax = read_proc_int("/proc/sys/net/core/rmem_max");
+
+      if (wmax > 0 && wmax < SEND_BUF_SIZE) {
+        RMW_UDS_LOG_WARN(
+          "net.core.wmem_max=%ld < requested SO_SNDBUF=%d. Large outbound "
+          "datagrams will fail with EMSGSIZE. Run: sudo sysctl -w "
+          "net.core.wmem_max=%d (or larger) and persist in /etc/sysctl.d/.",
+          wmax, SEND_BUF_SIZE, SEND_BUF_SIZE);
+      }
+      if (rmax > 0 && rmax < RECV_BUF_SIZE) {
+        RMW_UDS_LOG_WARN(
+          "net.core.rmem_max=%ld < requested SO_RCVBUF=%d. Subscribers may "
+          "drop large datagrams. Run: sudo sysctl -w net.core.rmem_max=%d "
+          "(or larger) and persist in /etc/sysctl.d/.",
+          rmax, RECV_BUF_SIZE, RECV_BUF_SIZE);
+      }
+    });
+}
+
 }  // namespace rmw_uds

rmw_unix_socket_cpp/src/transport.hpp

@@ -22,9 +22,11 @@ int create_bound_socket(const std::string & path);
 // Returns fd >= 0 on success, -1 on failure.
 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.
-// Returns true on success.
-bool send_to(
+SendResult send_to(
   int send_fd,
   const std::string & dest_path,
   const WireHeader & header,
@@ -49,6 +51,9 @@ void close_socket(int fd, const std::string & path);
 // Safe to call at startup to clean up after ungraceful shutdowns.
 void cleanup_orphan_socket_files(size_t domain_id);
 
+// One-shot per-process: warn if net.core.{w,r}mem_max < SO_{SND,RCV}BUF.
+void warn_if_sysctl_buffers_undersized();
+
 }  // namespace rmw_uds
 
 #endif  // RMW_UNIX_SOCKET_CPP__TRANSPORT_HPP_

rmw_unix_socket_cpp/test/test_rmw_qos.cpp

@@ -1,16 +1,22 @@
 #include "test_base.hpp"
 
+#include <cstdlib>
 #include <cstring>
 #include <thread>
 #include <chrono>
 
+#include <sys/socket.h>
+#include <unistd.h>
+
 #include "test_msgs/msg/basic_types.hpp"
 #include "test_msgs/srv/basic_types.hpp"
 
 #include "rmw/qos_profiles.h"
 #include "rosidl_typesupport_cpp/message_type_support.hpp"
 #include "rosidl_typesupport_cpp/service_type_support.hpp"
 
+#include "types.hpp"
+
 class QosTest : public RmwUdsNodeTest
 {
 protected:
@@ -134,6 +140,78 @@ TEST_F(QosTest, TransientLocalCacheDepthEnforced)
   auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
 }
 
+// --- Transport error propagation ---
+
+TEST_F(QosTest, PublishReturnsErrorOnEMSGSIZE)
+{
+  // Shrink SO_SNDBUF so any send hits EMSGSIZE — publish must return ERROR.
+  auto * ctx_impl = reinterpret_cast<rmw_uds::UdsContext *>(context.impl);
+  int small_buf = 2048;
+  ASSERT_EQ(
+    0,
+    setsockopt(
+      ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF,
+      &small_buf, sizeof(small_buf)));
+
+  auto qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE,
+    RMW_QOS_POLICY_DURABILITY_VOLATILE);
+
+  auto pub_opts = rmw_get_default_publisher_options();
+  auto * pub = rmw_create_publisher(node, ts, "/emsgsize", &qos, &pub_opts);
+  ASSERT_NE(nullptr, pub);
+
+  auto sub_opts = rmw_get_default_subscription_options();
+  auto * sub = rmw_create_subscription(node, ts, "/emsgsize", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  // 64 KB — well above SOCK_MIN_SNDBUF the kernel will clamp us to.
+  constexpr size_t big_size = 64 * 1024;
+  rmw_serialized_message_t serialized;
+  serialized.buffer_capacity = big_size;
+  serialized.buffer_length = big_size;
+  serialized.buffer = static_cast<uint8_t *>(std::malloc(big_size));
+  ASSERT_NE(nullptr, serialized.buffer);
+  std::memset(serialized.buffer, 0x42, big_size);
+  serialized.allocator = rcutils_get_default_allocator();
+
+  EXPECT_EQ(RMW_RET_ERROR, rmw_publish_serialized_message(pub, &serialized, nullptr));
+
+  std::free(serialized.buffer);
+  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.
+  auto qos = make_qos(
+    RMW_QOS_POLICY_RELIABILITY_RELIABLE,
+    RMW_QOS_POLICY_DURABILITY_VOLATILE);
+
+  auto pub_opts = rmw_get_default_publisher_options();
+  auto * pub = rmw_create_publisher(node, ts, "/peer_gone", &qos, &pub_opts);
+  ASSERT_NE(nullptr, pub);
+
+  auto sub_opts = rmw_get_default_subscription_options();
+  auto * sub = rmw_create_subscription(node, ts, "/peer_gone", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  // Warm pub's path cache, then unlink the sub's socket → next sendmsg = ENOENT.
+  test_msgs::msg::BasicTypes m;
+  m.int32_value = 1;
+  ASSERT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr));
+
+  auto * sub_impl = static_cast<rmw_uds::UdsSubscription *>(sub->data);
+  unlink(sub_impl->socket_path.c_str());
+
+  m.int32_value = 2;
+  EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr));
+
+  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(

rmw_unix_socket_cpp/test/test_transport.cpp

@@ -54,8 +54,10 @@ TEST_F(TransportTest, SendAndReceive)
   std::vector<uint8_t> payload = {1, 2, 3, 4, 5};
   send_hdr.payload_size = static_cast<uint32_t>(payload.size());
 
-  ASSERT_TRUE(rmw_uds::send_to(
-    send_fd, recv_path, send_hdr, payload.data(), payload.size()));
+  ASSERT_EQ(
+    rmw_uds::SendResult::Ok,
+    rmw_uds::send_to(
+      send_fd, recv_path, send_hdr, payload.data(), payload.size()));
 
   // Receive
   rmw_uds::WireHeader recv_hdr;
@@ -99,8 +101,10 @@ TEST_F(TransportTest, MultipleMessages)
     hdr.sequence_number = i;
     hdr.payload_size = sizeof(int);
     hdr.msg_type = 0;
-    ASSERT_TRUE(rmw_uds::send_to(
-      send_fd, path, hdr, reinterpret_cast<const uint8_t *>(&i), sizeof(i)));
+    ASSERT_EQ(
+      rmw_uds::SendResult::Ok,
+      rmw_uds::send_to(
+        send_fd, path, hdr, reinterpret_cast<const uint8_t *>(&i), sizeof(i)));
   }
 
   for (int i = 0; i < N; ++i) {