diff --git a/p2p/sentry/sentry_grpc_server.go b/p2p/sentry/sentry_grpc_server.go index 17f8a50b4f1..5e5feefc6f1 100644 --- a/p2p/sentry/sentry_grpc_server.go +++ b/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 diff --git a/p2p/sentry/sentry_grpc_server_test.go b/p2p/sentry/sentry_grpc_server_test.go index adad6d214fb..961cb977de5 100644 --- a/p2p/sentry/sentry_grpc_server_test.go +++ b/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") +}