fix(ros2): copy serialized buffers before callback return

Author: Nauyix

middlewares/ros2/include/ros2_subscription_wrapper.hpp

@@ -86,10 +86,9 @@ class SubscriptionManager {
   /**
    * Subscribe with a zero-copy (ABI v1.2) callback.
    *
-   * Fast path (no depth compression): retains the rclcpp::SerializedMessage
-   * shared_ptr and hands the recorder a pointer into its `buffer` along with
-   * a release function that drops the shared_ptr. No payload copy happens
-   * between rcl and the recorder's worker queue.
+   * Fast path (no depth compression): copies the rclcpp::SerializedMessage
+   * payload into plugin-owned storage before returning from the ROS callback,
+   * then hands the recorder that owned buffer plus a release function.
    *
    * Compression path (depth_compression.enabled): transparently falls back
    * to the v1.x copy semantics internally (the compressor allocates a new

middlewares/ros2/src/ros2_subscription_wrapper.cpp

@@ -247,21 +247,27 @@ bool SubscriptionManager::subscribe_v2(
           }
 #endif
 
-          // Zero-copy pass-through: hand the recorder a pointer into the rcl
-          // buffer and a holder that keeps the SerializedMessage alive. The
-          // release function deletes the heap-allocated shared_ptr holder,
-          // which in turn drops the last refcount (or whatever the refcount
-          // is at that point) on the SerializedMessage.
-          auto* holder = new std::shared_ptr<rclcpp::SerializedMessage>(msg);
+          // Copy the serialized payload into plugin-owned storage before
+          // returning from the ROS callback. Holding the rclcpp
+          // SerializedMessage in Axon's queues can exhaust DDS/rclcpp receive
+          // buffers when recorder batching or writer backlog retains many
+          // messages, throttling high-rate topics before they reach Axon.
           const auto& raw = msg->get_rcl_serialized_message();
+          auto* holder = new std::vector<uint8_t>();
+          if (raw.buffer != nullptr && raw.buffer_length > 0) {
+            holder->assign(raw.buffer, raw.buffer + raw.buffer_length);
+          }
+          if (holder->empty()) {
+            holder->resize(1);
+          }
           callback(
             topic_name,
             message_type,
-            raw.buffer,
+            holder->data(),
             raw.buffer_length,
             timestamp,
             +[](void* p) {
-              delete static_cast<std::shared_ptr<rclcpp::SerializedMessage>*>(p);
+              delete static_cast<std::vector<uint8_t>*>(p);
             },
             holder
           );