[performance] cherry-pick: p2p/sentry: don't duplicate SendMessageById across shared-store sentries (#21597)

p2p/sentry/sentry_grpc_server.go

@@ -1568,6 +1568,14 @@ func (ss *GrpcServer) SendMessageById(_ context.Context, inreq *sentryproto.Send
 		return reply, fmt.Errorf("msgcode not found for message Id: %s (peer protocol %d)", inreq.Data.Id, peerInfo.EthProtocol())
 	}
 
+	// With a shared PeerStore every sentry resolves the same PeerInfo, so only
+	// the sentry matching the peer's negotiated eth version may write — same
+	// rule as SendMessageToAll — or the peer receives one copy per sentry.
+	if protocolName, _ := ss.protocolForMessageID(inreq.Data.Id); protocolName == eth.ProtocolName &&
+		!protocolVersions.Contains(peerInfo.EthProtocol()) {
+		return reply, nil
+	}
+
 	ss.writePeer("[sentry] sendMessageById", peerInfo, inreq.Data.Id, msgcode, inreq.Data.Data, 0)
 	reply.Peers = []*typesproto.H512{inreq.PeerId}
 	return reply, nil

p2p/sentry/sentry_grpc_server_test.go

@@ -733,6 +733,8 @@ func newTestPeerInfoWithEth(t *testing.T) (*PeerInfo, [64]byte) {
 	t.Cleanup(rw.Close)
 
 	pi := NewPeerInfo(peer)
+	// Stop the per-peer worker goroutine started by NewPeerInfo.
+	t.Cleanup(pi.Close)
 	pi.SetEthRw(rw)
 	// Mark eth handshake as done so WaitForEth returns immediately.
 	pi.SetEthProtocol(direct.ETH68)
@@ -1462,3 +1464,80 @@ func TestGrpcServer_PeerEvents_ReplayFiltersByVersion(t *testing.T) {
 	require.Len(t, stream.events, 1, "PeerEvents replay must only emit Connect for the ETH68 peer")
 	require.Equal(t, sentryproto.PeerEvent_Connect, stream.events[0].EventId)
 }
+
+// countingMsgReadWriter counts WriteMsg calls; ReadMsg is never used by the
+// outbound write path under test.
+type countingMsgReadWriter struct {
+	mu     sync.Mutex
+	writes int
+}
+
+func (w *countingMsgReadWriter) ReadMsg() (p2p.Msg, error) {
+	return p2p.Msg{}, io.EOF
+}
+
+func (w *countingMsgReadWriter) WriteMsg(msg p2p.Msg) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.writes++
+	return nil
+}
+
+func (w *countingMsgReadWriter) count() int {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.writes
+}
+
+// TestGrpcServer_SendMessageById_SharedStore_NoDuplicateWrites: in shared
+// p2p.Server mode there is one GrpcServer per eth version, all backed by one
+// PeerStore, and clients (e.g. the txpool) fan SendMessageById out across
+// every sentry. The message must reach the peer once — via the sentry whose
+// version the peer negotiated — not once per sentry.
+func TestGrpcServer_SendMessageById_SharedStore_NoDuplicateWrites(t *testing.T) {
+	shared := NewPeerStore()
+	versions := []uint{direct.ETH69, direct.ETH70, direct.ETH71}
+	servers := make([]*GrpcServer, 0, len(versions))
+	for _, v := range versions {
+		ss := &GrpcServer{
+			statusReady: make(chan struct{}),
+			ethVersion:  v,
+			logger:      log.New(),
+			Protocols: []p2p.Protocol{{
+				Name:      eth.ProtocolName,
+				Version:   v,
+				FromProto: eth.FromProto[v],
+			}},
+		}
+		ss.peers.Store(NewPeerStore())
+		ss.SetSharedPeerStore(shared)
+		servers = append(servers, ss)
+	}
+
+	pi, peerID := newTestPeerInfoWithEth(t)
+	rw := &countingMsgReadWriter{}
+	pi.SetEthRw(rw)
+	pi.SetEthProtocol(direct.ETH70)
+	store := servers[0].peers.Load()
+	store.mu.Lock()
+	store.peers[peerID] = pi
+	store.mu.Unlock()
+
+	req := &sentryproto.SendMessageByIdRequest{
+		PeerId: gointerfaces.ConvertHashToH512(peerID),
+		Data: &sentryproto.OutboundMessageData{
+			Id:   sentryproto.MessageId_NEW_POOLED_TRANSACTION_HASHES_68,
+			Data: []byte{0xc0},
+		},
+	}
+	for _, ss := range servers {
+		_, err := ss.SendMessageById(context.Background(), req)
+		require.NoError(t, err)
+	}
+
+	// Writes happen on the peer's async worker; wait for the first, then
+	// allow a settle window to catch duplicates from the other sentries.
+	require.Eventually(t, func() bool { return rw.count() >= 1 }, time.Second, 5*time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
+	require.Equal(t, 1, rw.count(), "peer negotiated eth/70: only the eth/70 sentry must write")
+}