fix: duplicate RDMA ACK fd before acquiring send stream

rdma_comm_sender::get_send_stream() receives a borrowed ACK fd from the
replication layer. Duplicate it before wrapping it in rdma_comm::unique_fd
so rdma_comm can take ownership without closing the caller-owned fd.

Also document the borrowed-fd ownership contract in rdma_sender_base.

Author: umegane

src/limestone/rdma/rdma_comm/rdma_comm_sender.cpp

@@ -16,6 +16,11 @@
 #include <rdma/rdma_comm_sender.h>
 #include <rdma/rdma_comm_send_stream.h>
 
+#include <cerrno>
+#include <cstring>
+#include <string>
+#include <unistd.h>
+
 #include <rdma_comm/unique_fd.h>
 
 namespace limestone::replication {
@@ -33,9 +38,14 @@ rdma_sender_base::operation_result rdma_comm_sender::initialize(
 rdma_sender_base::stream_acquire_result rdma_comm_sender::get_send_stream(
         std::uint16_t channel_id,
         int           ack_fd) noexcept {
+    int owned_ack_fd = ::dup(ack_fd);
+    if (owned_ack_fd < 0) {
+        return {{false, std::string{"dup() failed for RDMA ACK fd: "} + std::strerror(errno)}, nullptr};
+    }
+
     auto r = sender_.get_send_stream(
         channel_id,
-        rdma::communication::unique_fd{ack_fd});
+        rdma::communication::unique_fd{owned_ack_fd});
     if (! r.status.success || ! r.stream) {
         return {{false, r.status.error_message}, nullptr};
     }

src/limestone/rdma/rdma_sender_base.h

@@ -66,7 +66,8 @@ class rdma_sender_base {
     /**
      * @brief Acquire a send stream for the given logical channel.
      * @param channel_id Logical channel identifier.
-     * @param ack_fd     File descriptor used to receive acknowledgements.
+     * @param ack_fd     Borrowed file descriptor used to receive acknowledgements.
+     *                   Implementations must duplicate it before taking ownership.
      * @return stream_acquire_result; stream is non-null on success.
      */
     [[nodiscard]] virtual stream_acquire_result get_send_stream(