test(replication): add rdma blob error-path coverage

Add regression tests for incomplete RDMA BLOB bodies and BLOB size mismatches.
These cases leave partial BLOB files on disk but must not complete the LOG_ENTRY
message, so no WAL entry can reference the incomplete BLOB.

Also cover multiple BLOB LOG_ENTRY messages in one RDMA receiver stream.

Author: umegane

test/limestone/replication/rdma_log_entries_parser_test.cpp

@@ -1,6 +1,7 @@
 #include "replication/rdma_log_entries_parser.h"
 
 #include <boost/filesystem.hpp>
+#include <arpa/inet.h>
 #include <fstream>
 #include <iterator>
 #include <string>
@@ -536,4 +537,96 @@ TEST(rdma_log_entries_parser_test, partial_blob_body_keeps_reading_until_remaini
     boost::filesystem::remove_all(receiver_location);
 }
 
+TEST(rdma_log_entries_parser_test, incomplete_blob_body_keeps_partial_file_and_message_incomplete) {
+    // Simulate a transfer that stops before all BLOB body bytes arrive.  The
+    // parser may leave the partially written BLOB file on disk, but the
+    // LOG_ENTRY message must stay incomplete so no WAL entry can reference it.
+    static constexpr const char* sender_location = "/tmp/rdma_log_entries_parser_incomplete_sender_test";
+    static constexpr const char* receiver_location = "/tmp/rdma_log_entries_parser_incomplete_receiver_test";
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+
+    limestone::api::configuration sender_conf{};
+    sender_conf.set_data_location(sender_location);
+    limestone::api::datastore_test sender_datastore{sender_conf};
+
+    limestone::api::configuration receiver_conf{};
+    receiver_conf.set_data_location(receiver_location);
+    limestone::api::datastore_test receiver_datastore{receiver_conf};
+
+    constexpr limestone::api::blob_id_type blob_id = 6001;
+    std::string const blob_body = "incomplete-blob-body";
+    auto sender_blob_path = sender_datastore.get_blob_file(blob_id).path();
+    boost::filesystem::create_directories(sender_blob_path.parent_path());
+    std::ofstream(sender_blob_path.string(), std::ios::binary) << blob_body;
+
+    message_log_entries original{905};
+    original.add_normal_with_blob(3, "key", "value", {4, 5}, {blob_id});
+
+    blob_socket_io io("", sender_datastore);
+    original.send_body(io);
+    std::string body = io.get_out_string();
+
+    auto receiver_blob_path = receiver_datastore.get_blob_file(blob_id).path();
+    auto const truncated_size = body.size() - 3;
+
+    rdma_log_entries_parser parser{receiver_datastore};
+    EXPECT_EQ(parser.consume(std::string_view{body}.substr(0, truncated_size)), truncated_size);
+    EXPECT_FALSE(parser.complete());
+    EXPECT_EQ(parser.get_status(), rdma_log_entries_parser::status::reading);
+    EXPECT_TRUE(boost::filesystem::exists(receiver_blob_path));
+    EXPECT_THROW([[maybe_unused]] auto msg = parser.take_message(), std::logic_error);
+
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+}
+
+TEST(rdma_log_entries_parser_test, blob_size_mismatch_larger_than_payload_leaves_partial_file_and_incomplete_message) {
+    // Corrupt the encoded BLOB size so it is larger than the bytes actually
+    // present in the payload.  This models a size mismatch / missing trailing
+    // chunk: the parser must keep waiting instead of completing the message
+    // with a reference to an incomplete BLOB.
+    static constexpr const char* sender_location = "/tmp/rdma_log_entries_parser_size_mismatch_sender_test";
+    static constexpr const char* receiver_location = "/tmp/rdma_log_entries_parser_size_mismatch_receiver_test";
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+
+    limestone::api::configuration sender_conf{};
+    sender_conf.set_data_location(sender_location);
+    limestone::api::datastore_test sender_datastore{sender_conf};
+
+    limestone::api::configuration receiver_conf{};
+    receiver_conf.set_data_location(receiver_location);
+    limestone::api::datastore_test receiver_datastore{receiver_conf};
+
+    constexpr limestone::api::blob_id_type blob_id = 6002;
+    std::string const blob_body = "blob-size-mismatch";
+    auto sender_blob_path = sender_datastore.get_blob_file(blob_id).path();
+    boost::filesystem::create_directories(sender_blob_path.parent_path());
+    std::ofstream(sender_blob_path.string(), std::ios::binary) << blob_body;
+
+    message_log_entries original{906};
+    original.add_normal_with_blob(3, "key", "value", {4, 5}, {blob_id});
+
+    blob_socket_io io("", sender_datastore);
+    original.send_body(io);
+    std::string body = io.get_out_string();
+
+    auto const blob_size_offset = body.size() - 1 - blob_body.size() - sizeof(std::uint32_t);
+    auto encoded_size = htonl(static_cast<std::uint32_t>(blob_body.size() + 5));
+    std::memcpy(body.data() + static_cast<std::ptrdiff_t>(blob_size_offset), &encoded_size, sizeof(encoded_size));
+
+    auto receiver_blob_path = receiver_datastore.get_blob_file(blob_id).path();
+
+    rdma_log_entries_parser parser{receiver_datastore};
+    EXPECT_EQ(parser.consume(body), body.size());
+    EXPECT_FALSE(parser.complete());
+    EXPECT_EQ(parser.get_status(), rdma_log_entries_parser::status::reading);
+    EXPECT_TRUE(boost::filesystem::exists(receiver_blob_path));
+    EXPECT_THROW([[maybe_unused]] auto msg = parser.take_message(), std::logic_error);
+
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+}
+
 }  // namespace limestone::testing

test/limestone/replication/rdma_log_entries_receiver_test.cpp

@@ -183,6 +183,81 @@ TEST(rdma_log_entries_receiver_test, streams_blob_message_split_across_inputs) {
     boost::filesystem::remove_all(receiver_location);
 }
 
+TEST(rdma_log_entries_receiver_test, parses_multiple_blob_messages_in_one_session) {
+    // A persistent session may send more than one LOG_ENTRY message containing
+    // BLOBs.  Feed two serialized BLOB messages as one RDMA byte stream, split
+    // across arbitrary consume() boundaries, and verify that both messages and
+    // both replica BLOB files are completed independently.
+    static constexpr const char* sender_location = "/tmp/rdma_log_entries_receiver_multi_blob_sender_test";
+    static constexpr const char* receiver_location = "/tmp/rdma_log_entries_receiver_multi_blob_receiver_test";
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+
+    limestone::api::configuration sender_conf{};
+    sender_conf.set_data_location(sender_location);
+    limestone::api::datastore_test sender_datastore{sender_conf};
+
+    limestone::api::configuration receiver_conf{};
+    receiver_conf.set_data_location(receiver_location);
+    limestone::api::datastore_test receiver_datastore{receiver_conf};
+
+    constexpr limestone::api::blob_id_type first_blob_id = 9877;
+    constexpr limestone::api::blob_id_type second_blob_id = 9878;
+    std::string const first_blob_body = "first-session-blob";
+    std::string const second_blob_body = "second-session-blob";
+
+    auto first_sender_blob_path = sender_datastore.get_blob_file(first_blob_id).path();
+    boost::filesystem::create_directories(first_sender_blob_path.parent_path());
+    std::ofstream(first_sender_blob_path.string(), std::ios::binary) << first_blob_body;
+
+    auto second_sender_blob_path = sender_datastore.get_blob_file(second_blob_id).path();
+    boost::filesystem::create_directories(second_sender_blob_path.parent_path());
+    std::ofstream(second_sender_blob_path.string(), std::ios::binary) << second_blob_body;
+
+    message_log_entries first{1202};
+    first.set_session_begin_flag(true);
+    first.add_normal_with_blob(3, "blob-key-1", "blob-value-1", {4, 5}, {first_blob_id});
+
+    message_log_entries second{1202};
+    second.set_session_end_flag(true);
+    second.add_normal_with_blob(4, "blob-key-2", "blob-value-2", {4, 6}, {second_blob_id});
+
+    std::string bytes = serialize_message_with_blobs(first, sender_datastore)
+                      + serialize_message_with_blobs(second, sender_datastore);
+
+    rdma_log_entries_receiver receiver{receiver_datastore};
+    std::size_t consumed = 0;
+    while (consumed < bytes.size()) {
+        std::size_t chunk = consumed % 2 == 0 ? 7u : 3u;
+        chunk = std::min(chunk, bytes.size() - consumed);
+        consumed += receiver.consume(std::string_view{bytes}.substr(consumed, chunk));
+    }
+
+    ASSERT_EQ(receiver.message_count(), 2u);
+
+    auto first_received = receiver.take_message();
+    EXPECT_EQ(first_received->get_epoch_id(), 1202);
+    EXPECT_TRUE(first_received->has_session_begin_flag());
+    ASSERT_EQ(first_received->get_entries().size(), 1u);
+    EXPECT_EQ(first_received->get_entries()[0].blob_ids,
+            std::vector<limestone::api::blob_id_type>{first_blob_id});
+
+    auto second_received = receiver.take_message();
+    EXPECT_EQ(second_received->get_epoch_id(), 1202);
+    EXPECT_TRUE(second_received->has_session_end_flag());
+    ASSERT_EQ(second_received->get_entries().size(), 1u);
+    EXPECT_EQ(second_received->get_entries()[0].blob_ids,
+            std::vector<limestone::api::blob_id_type>{second_blob_id});
+
+    auto first_receiver_blob_path = receiver_datastore.get_blob_file(first_blob_id).path();
+    auto second_receiver_blob_path = receiver_datastore.get_blob_file(second_blob_id).path();
+    EXPECT_EQ(read_file(first_receiver_blob_path), first_blob_body);
+    EXPECT_EQ(read_file(second_receiver_blob_path), second_blob_body);
+
+    boost::filesystem::remove_all(sender_location);
+    boost::filesystem::remove_all(receiver_location);
+}
+
 TEST(rdma_log_entries_receiver_test, leaves_partial_next_message_in_progress_after_completed_message) {
     // If the input ends after the next message type byte, the previous message
     // should still be available while the receiver remembers that a new message