[TENT] Retire stale endpoint on peer reconnect bootstrap

mooncake-transfer-engine/tent/include/tent/transport/rdma/endpoint.h

@@ -192,6 +192,8 @@ class RdmaEndPoint : public std::enable_shared_from_this<RdmaEndPoint> {
     void repostAllNotifyRecvs();
 
    private:
+    friend class EndpointTestAccess;
+
     std::atomic<EndPointStatus> status_;
     RdmaContext* context_;
     EndPointParams* params_;

mooncake-transfer-engine/tent/src/transport/rdma/endpoint.cpp

@@ -576,15 +576,19 @@ Status RdmaEndPoint::accept(const BootstrapDesc& peer_desc,
             local_desc.notify_qp_num = notifyQpNum();
             return mooncake::tent::Status::OK();
         }
-        // Endpoint already connected to a different peer - reject the request
-        // instead of resetting. Endpoints have unidirectional lifecycle and
-        // are never reset or reused. The caller should create a new endpoint.
-        LOG(ERROR)
-            << "Endpoint already established with " << peer_nic_name_ << " of "
-            << peer_server_name_
-            << ", cannot accept new connection (unidirectional lifecycle)";
+        // The bootstrap does not match the established connection: the peer
+        // discarded its endpoint (eviction or failure) and came back with new
+        // QPs, so the local QPs now point at QPs that no longer exist.
+        // Endpoints have unidirectional lifecycle and are never reset, so
+        // retire this one. The caller drops retiring endpoints from the store,
+        // and the peer's next bootstrap gets a newly created endpoint.
+        LOG(WARNING) << "Endpoint already established with " << peer_nic_name_
+                     << " of " << peer_server_name_
+                     << ", retiring it for the new connection (unidirectional "
+                        "lifecycle)";
+        beginDestroyNoLock();
         return mooncake::tent::Status::InternalError(
-            "Endpoint already connected to different peer" LOC_MARK);
+            "Endpoint retired for reconnection from peer" LOC_MARK);
     }
     if (status_.load(std::memory_order_relaxed) != EP_HANDSHAKING) {
         LOG(ERROR) << "Endpoint not in handshaking state: "

mooncake-transfer-engine/tent/tests/endpoint_lifecycle_test.cpp

@@ -16,11 +16,29 @@
 
 #include <memory>
 
+#include "tent/common/utils/string_builder.h"
 #include "tent/transport/rdma/endpoint.h"
 #include "tent/transport/rdma/slice.h"
 
 namespace mooncake {
 namespace tent {
+
+class EndpointTestAccess {
+   public:
+    // Puts a context-less endpoint into the state a completed bootstrap
+    // leaves behind, so accept() can be driven without an RDMA device.
+    static void markConnected(RdmaEndPoint& endpoint,
+                              const std::string& peer_server_name,
+                              const std::string& peer_nic_name,
+                              const std::vector<uint32_t>& peer_qp_num_list) {
+        endpoint.peer_server_name_ = peer_server_name;
+        endpoint.peer_nic_name_ = peer_nic_name;
+        endpoint.peer_qp_num_list_ = peer_qp_num_list;
+        endpoint.status_.store(RdmaEndPoint::EP_READY,
+                               std::memory_order_relaxed);
+    }
+};
+
 namespace {
 
 TEST(EndpointLifecycleTest, DefaultConstructedEndpointOwnsNoResources) {
@@ -152,6 +170,23 @@ TEST(EndpointLifecycleTest, SliceWeakPtrResetClearsReference) {
     EXPECT_NE(endpoint, nullptr);
 }
 
+TEST(EndpointLifecycleTest, BootstrapWithNewPeerQpsRetiresEstablishedEndpoint) {
+    // The peer dropped its endpoint (store eviction or a transfer failure)
+    // and bootstraps again with a fresh QP set. The established endpoint now
+    // points at QPs that no longer exist, so it must retire instead of
+    // staying EP_READY and rejecting every later bootstrap from that peer.
+    RdmaEndPoint endpoint;
+    EndpointTestAccess::markConnected(endpoint, "10.0.0.1:12345", "mlx5_0",
+                                      {100, 101});
+
+    BootstrapDesc peer_desc, local_desc;
+    peer_desc.local_nic_path = MakeNicPath("10.0.0.1:12345", "mlx5_0");
+    peer_desc.qp_num = {200, 201};
+
+    EXPECT_FALSE(endpoint.accept(peer_desc, local_desc).ok());
+    EXPECT_EQ(endpoint.status(), RdmaEndPoint::EP_DESTROYING);
+}
+
 TEST(EndpointLifecycleTest, ExternalOwnerCanReleaseAfterExplicitDeconstruct) {
     auto endpoint = std::make_shared<RdmaEndPoint>();
     std::weak_ptr<RdmaEndPoint> weak = endpoint;