fix _read_buf race between PollCq and OnNewMessages in RDMA server

[bzs1118]

The server-side RDMA socket's _read_buf is accessed by two independent bthreads: PollCq (CQ socket) writes RDMA data via HandleCompletion, and OnNewMessages(main socket) reads TCP data for handshake / fallback. Since IOBuf is not thread-safe, concurrent access corrupts internal state and causes intermittent core dumps.

Three fixes:

1. Switch edge trigger to OnNewDataFromTcpin ALL ExecuteServerHandshake end paths (ESTABLISHED + 5 failure paths). OnNewDataFromTcpchecks the RDMA state: in ESTABLISHED it only reads 1 byte for EOF detection without touching _read_buf; in FALLBACK_TCP it delegates to OnNewMessages for TCP data. This prevents post-handshake races.

2. Guard HandleCompletion(IBV_WC_RECV) with a state check: skip writing to _read_buf and re-post the recv WR if the state is not ESTABLISHED. This prevents races during the handshake (after BringUpQp puts the QP into RTS, the client may start sending RDMA data before the server finishes processing the ACK).

3. Remove the source->size() > HELLO_ACK_LEN check in Phase 2. When a client falls back to TCP, the 4-byte ACK and the first RPC request may arrive in the same readv() call. Use cutn() to drain the 4-byte ACK and let remaining data be processed by other parsers, matching FallbackServerHandshake's behavior.

4. Return NOT_ENOUGH_DATA (not TRY_OTHERS) from the ESTABLISHED path so OnNewMessages stops processing _read_buf before PollCq starts writing.

5. Clear _read_buf before transitioning to ESTABLISHED so that residual TCP data cannot become a prefix of the RDMA recv stream (HandleCompletion appends to _read_buf, not overwrites). The clear is safe because HandleCompletion only writes after seeing ESTABLISHED (acquire), which is stored (release) strictly after the clear.

6. Restore edge trigger in RdmaTransport::Reset() based on CreatedByConnect(): OnNewDataFromTcp for client-side sockets, OnNewMessages for server-side sockets, matching the logic in Init().

7. Add Transport::ShouldStopReading() virtual method (default false), overridden by
   RdmaTransport to return true when _rdma_state == RDMA_ON. OnNewMessages checks this after ProcessNewMessage returns and exits immediately, preventing it from calling DoRead again on _read_buf after the edge trigger has been switched.

8. Guard ProcessNewMessage in PollCq with bytes > 0: when bytes == 0 (IBV_WC_SEND completions, or IBV_WC_RECV dropped during handshake), skip ProcessNewMessage entirely. This prevents PollCq from calling CutInputMessage on _read_buf while OnNewMessages is driving the handshake on the same _read_buf.

What problem does this PR solve?

Issue Number: #3479

Problem Summary:

What is changed and the side effects?

Changed:

Side effects:

• Performance effects: HandleCompletion adds one atomic load per recv completion (negligible)

• ShouldStopReading() adds one virtual call per ProcessNewMessage iteration (negligible);

• Breaking backward compatibility: No. The OnNewDataFromTcp behavior in FALLBACK_TCP state is unchanged (delegates to OnNewMessages). The removed > HELLO_ACK_LEN check aligns with FallbackServerHandshake's existing behavior. ShouldStopReading() defaults to false for non-RDMA transports.

---

Check List:

• Please make sure your changes are compilable.
• When providing us with a new feature, it is best to add related tests.
• Please follow Contributor Covenant Code of Conduct.

[Copilot]

Pull request overview

This PR addresses an RDMA server crash caused by concurrent access to Socket::_read_buf from two bthreads (PollCq via HandleCompletion() and the TCP-driven OnNewMessages handshake/fallback path). It does so by switching the edge-trigger handler after handshake, adding a state gate in RDMA recv completion handling, and relaxing overly-strict ACK parsing so TCP-fallback clients can coalesce ACK + first request.

Changes:

• Switch server-side edge trigger to RdmaEndpoint::OnNewDataFromTcp on all handshake terminal paths, and change the ESTABLISHED return to PARSE_ERROR_NOT_ENOUGH_DATA to stop OnNewMessages from further parsing.
• In HandleCompletion(IBV_WC_RECV), skip touching _read_buf unless state is ESTABLISHED.
• In RdmaTransport::Reset(), restore the edge-trigger handler (added in this PR).

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
src/brpc/rdma/rdma_endpoint.cpp	Adjusts server handshake terminal behavior/returns and adds a state gate in RDMA recv completion to avoid concurrent _read_buf mutation.
src/brpc/rdma_transport.cpp	Resets the transport edge-trigger callback during Reset() to support re-handshake behavior.

Suppressed comments (2)

src/brpc/rdma/rdma_endpoint.cpp:935

• This branch intentionally drops RDMA payloads received before the server reaches ESTABLISHED. The issue description notes this can happen after BringUpQp (client may start sending RDMA data before server finishes processing the ACK), so dropping here risks losing the first real RPC bytes and causing timeouts/hangs. Consider buffering these bytes until the handshake finalizes, or preventing the main socket from touching _read_buf during S_ACK_WAIT so PollCq can safely append without loss.

            // Don't write to _read_buf until the handshake is fully done
            // (ESTABLISHED). During the handshake (S_ACK_WAIT etc.), the
            // main socket's OnNewMessages is driving the handshake via
            // _read_buf; PollCq writing to _read_buf concurrently corrupts
            // the IOBuf (non-thread-safe).
            if (_state.load(butil::memory_order_acquire) != ESTABLISHED) {
                LOG(WARNING) << "RDMA recv completion in non-ESTABLISHED state "
                             << GetStateStr() << ", drop "
                             << wc.byte_len << " bytes from "
                             << _socket->description();

src/brpc/rdma/rdma_endpoint.cpp:936

• In the non-ESTABLISHED recv-completion path, PostRecv() failures are ignored, which can leave the receive queue un-posted and stall the connection. Also, a plain LOG(WARNING) here can spam logs if the peer sends early/invalid RDMA traffic. It’s safer to (a) check PostRecv() and propagate failure and (b) rate-limit the warning log.

                LOG(WARNING) << "RDMA recv completion in non-ESTABLISHED state "
                             << GetStateStr() << ", drop "
                             << wc.byte_len << " bytes from "
                             << _socket->description();
                PostRecv(1, zerocopy);

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

[bzs1118]

addressed this by adding a Transport::ShouldStopReading() virtual mechanism