From 010a3e11204a4b664101cf56e1469796fed4633c Mon Sep 17 00:00:00 2001
From: Lijin Xiong <lijin.xiong@zstack.io>
Date: Mon, 27 Jul 2026 13:14:00 +0800
Subject: [PATCH] Reject stale CQ callbacks after socket revival

A CQ callback can remain queued across Reset() and run after the main
socket has been revived with a different CQ. Verify that the callback's
CQ SocketId still matches the endpoint before accessing RDMA resources.
Add a regression test covering a stale callback from an older
generation.

Signed-off-by: Lijin Xiong <legion.xiong@gmail.com>
---
 src/brpc/rdma/rdma_endpoint.cpp |  5 +++++
 test/brpc_rdma_unittest.cpp     | 34 +++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp
index ca6f3929af..d0fad7536f 100644
--- a/src/brpc/rdma/rdma_endpoint.cpp
+++ b/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);
 
diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp
index e30ae09f35..0435641e74 100644
--- a/test/brpc_rdma_unittest.cpp
+++ b/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();