Reject stale CQ callbacks after socket revival

src/brpc/rdma/rdma_endpoint.cpp

@@ -1434,6 +1434,11 @@ void RdmaEndpoint::PollCq(Socket* m) {
     if (Socket::Address(ep->_socket->id(), &s) < 0) {
         return;
     }
+    // A queued callback may outlive Reset() and see the main Socket after
+    // it has been revived with another CQ.
+    if (m->id() != ep->_cq_sid) {
+        return;
+    }
     auto* rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
     CHECK(ep == rdma_transport->_rdma_ep);
 

test/brpc_rdma_unittest.cpp

@@ -186,6 +186,40 @@ class RdmaRpcTest : public RdmaTest,
     int _saved_handshake_version = 2;
 };
 
+TEST_F(RdmaTest, stale_cq_callback_does_not_poll_new_generation) {
+    SocketOptions main_options;
+    main_options.socket_mode = SOCKET_MODE_RDMA;
+    SocketId main_sid;
+    ASSERT_EQ(0, Socket::Create(main_options, &main_sid));
+
+    SocketUniquePtr main_socket;
+    ASSERT_EQ(0, Socket::Address(main_sid, &main_socket));
+    RdmaTransport* transport =
+        static_cast<RdmaTransport*>(main_socket->_transport.get());
+    rdma::RdmaEndpoint* ep = transport->_rdma_ep;
+
+    SocketOptions cq_options;
+    cq_options.user = ep;
+    SocketId stale_cq_sid;
+    SocketId current_cq_sid;
+    ASSERT_EQ(0, Socket::Create(cq_options, &stale_cq_sid));
+    ASSERT_EQ(0, Socket::Create(cq_options, &current_cq_sid));
+
+    SocketUniquePtr stale_cq_socket;
+    SocketUniquePtr current_cq_socket;
+    ASSERT_EQ(0, Socket::Address(stale_cq_sid, &stale_cq_socket));
+    ASSERT_EQ(0, Socket::Address(current_cq_sid, &current_cq_socket));
+    ep->_cq_sid = current_cq_sid;
+
+    rdma::RdmaEndpoint::PollCq(stale_cq_socket.get());
+
+    stale_cq_socket->_user = NULL;
+    current_cq_socket->_user = NULL;
+    stale_cq_socket->SetFailed();
+    current_cq_socket->SetFailed();
+    main_socket->SetFailed();
+}
+
 TEST_F(RdmaTest, client_close_before_hello_send) {
     StartServer();