test: cover rmw_take_sequence mid-batch deserialize failure

Inject [good, corrupt, good] datagrams into a subscription socket and assert
take_sequence yields two contiguous messages (size == *taken == 2) with no
hole and nothing lost. Fails before the cursor fix (data[1] was a hole and
the second good message, written at data[2], was dropped).

Author: benaliabderrahmane

rmw_unix_socket_cpp/test/test_rmw_pub_sub.cpp

@@ -8,6 +8,16 @@
 #include "rmw/qos_profiles.h"
 #include "rosidl_typesupport_cpp/message_type_support.hpp"
 
+#include <chrono>
+#include <thread>
+#include <vector>
+
+#include <unistd.h>
+
+#include "../src/types.hpp"
+#include "../src/transport.hpp"
+#include "../src/serialization.hpp"
+
 class PubSubTest : public RmwUdsNodeTest
 {
 protected:
@@ -200,3 +210,80 @@ TEST_F(PubSubTest, StringMessages)
   ASSERT_TRUE(taken);
   EXPECT_EQ("Hello from Unix sockets!", recv_msg.string_value);
 }
+
+// Regression for the rmw_take_sequence cursor bug: a deserialize failure in the
+// middle of a batch must not leave an uninitialized hole or miscount size — each
+// success is written at the *taken cursor, not the loop index. We inject
+// [good, corrupt, good] straight into the subscription's datagram socket; a
+// single sender over SOCK_DGRAM preserves datagram order, so the corrupt one
+// lands between the two good ones. Before the fix the second good message was
+// written at data[2] and lost, while size == *taken == 2 exposed data[1] (the
+// hole left by the failed deserialize) to the consumer as a valid message.
+TEST_F(PubSubTest, TakeSequenceSkipsMidBatchCorruptContiguously)
+{
+  auto sub_opts = rmw_get_default_subscription_options();
+  sub = rmw_create_subscription(node, ts, "/take_seq_corrupt", &qos, &sub_opts);
+  ASSERT_NE(nullptr, sub);
+
+  auto * sub_data = static_cast<rmw_uds::UdsSubscription *>(sub->data);
+  ASSERT_FALSE(sub_data->socket_path.empty());
+  ASSERT_NE(nullptr, sub_data->callbacks);
+
+  // A valid serialized BasicTypes payload (deserializes successfully).
+  test_msgs::msg::BasicTypes good;
+  good.int32_value = 4242;
+  good.bool_value = true;
+  std::vector<uint8_t> good_payload;
+  ASSERT_TRUE(rmw_uds::serialize(&good, sub_data->callbacks, good_payload));
+
+  int send_fd = rmw_uds::create_send_socket();
+  ASSERT_GE(send_fd, 0);
+
+  auto inject = [&](const uint8_t * data, size_t len) {
+    rmw_uds::WireHeader hdr;
+    std::memset(&hdr, 0, sizeof(hdr));
+    hdr.payload_size = static_cast<uint32_t>(len);
+    hdr.msg_type = 0;  // topic message
+    ASSERT_TRUE(rmw_uds::send_to(send_fd, sub_data->socket_path, hdr, data, len));
+  };
+
+  // good, corrupt (far too short to deserialize as BasicTypes), good.
+  const uint8_t corrupt[4] = {0xDE, 0xAD, 0xBE, 0xEF};
+  inject(good_payload.data(), good_payload.size());
+  inject(corrupt, sizeof(corrupt));
+  inject(good_payload.data(), good_payload.size());
+
+  // Datagrams are buffered by the kernel on send; a short pause is defensive.
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+  // Take all three at once: the corrupt one is skipped, leaving two contiguous
+  // good messages with size == *taken == 2 (no hole, nothing lost).
+  test_msgs::msg::BasicTypes out[3];
+  void * out_ptrs[3] = {&out[0], &out[1], &out[2]};
+  rmw_message_sequence_t seq;
+  seq.data = out_ptrs;
+  seq.size = 0;
+  seq.capacity = 3;
+  seq.allocator = nullptr;
+
+  rmw_message_info_t infos[3];
+  std::memset(infos, 0, sizeof(infos));
+  rmw_message_info_sequence_t info_seq;
+  info_seq.data = infos;
+  info_seq.size = 0;
+  info_seq.capacity = 3;
+  info_seq.allocator = nullptr;
+
+  size_t taken = 99;
+  EXPECT_EQ(
+    RMW_RET_OK,
+    rmw_take_sequence(sub, 3, &seq, &info_seq, &taken, nullptr));
+
+  EXPECT_EQ(2u, taken);
+  EXPECT_EQ(2u, seq.size);
+  EXPECT_EQ(2u, info_seq.size);
+  EXPECT_EQ(4242, out[0].int32_value);
+  EXPECT_EQ(4242, out[1].int32_value);  // the bug left this a hole and lost it
+
+  close(send_fd);
+}