diff --git a/node/direct/sentry_client.go b/node/direct/sentry_client.go
index 0380fb80bef..b691840d39e 100644
--- a/node/direct/sentry_client.go
+++ b/node/direct/sentry_client.go
@@ -19,11 +19,9 @@ package direct
import (
"context"
"fmt"
- "io"
"sync"
"google.golang.org/grpc"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/emptypb"
"github.com/erigontech/erigon/node/gointerfaces/sentryproto"
@@ -242,86 +240,28 @@ func (c *SentryClientDirect) PeerById(ctx context.Context, in *sentryproto.PeerB
return c.server.PeerById(ctx, in)
}
-// -- start Messages
-
func (c *SentryClientDirect) Messages(ctx context.Context, in *sentryproto.MessagesRequest, opts ...grpc.CallOption) (sentryproto.Sentry_MessagesClient, error) {
allProtocols := append([]sentryproto.Protocol{c.protocol}, c.sideProtocols...)
in = &sentryproto.MessagesRequest{
Ids: filterIds(in.Ids, allProtocols),
}
- ch := make(chan *inboundMessageReply, libsentry.MessagesQueueSize)
- streamServer := &SentryMessagesStreamS{ch: ch, ctx: ctx}
+ ch := make(chan libsentry.StreamReply[*sentryproto.InboundMessage], libsentry.MessagesQueueSize)
+ streamServer := &libsentry.SentryStreamS[*sentryproto.InboundMessage]{Ch: ch, Ctx: ctx}
go func() {
defer close(ch)
streamServer.Err(c.server.Messages(in, streamServer))
}()
- return &SentryMessagesStreamC{ch: ch, ctx: ctx}, nil
-}
-
-type inboundMessageReply struct {
- r *sentryproto.InboundMessage
- err error
-}
-
-// SentryMessagesStreamS implements proto_sentryproto.Sentry_ReceiveMessagesServer
-type SentryMessagesStreamS struct {
- ch chan *inboundMessageReply
- ctx context.Context
- grpc.ServerStream
-}
-
-func (s *SentryMessagesStreamS) Send(m *sentryproto.InboundMessage) error {
- s.ch <- &inboundMessageReply{r: m}
- libsentry.EvictOldestIfHalfFull(s.ch)
- return nil
-}
-
-func (s *SentryMessagesStreamS) Context() context.Context { return s.ctx }
-
-func (s *SentryMessagesStreamS) Err(err error) {
- if err == nil {
- return
- }
- s.ch <- &inboundMessageReply{err: err}
+ return &libsentry.SentryStreamC[*sentryproto.InboundMessage]{Ch: ch, Ctx: ctx}, nil
}
-type SentryMessagesStreamC struct {
- ch chan *inboundMessageReply
- ctx context.Context
- grpc.ClientStream
-}
-
-func (c *SentryMessagesStreamC) Recv() (*sentryproto.InboundMessage, error) {
- m, ok := <-c.ch
- if !ok || m == nil {
- return nil, io.EOF
- }
- return m.r, m.err
-}
-
-func (c *SentryMessagesStreamC) Context() context.Context { return c.ctx }
-
-func (c *SentryMessagesStreamC) RecvMsg(anyMessage any) error {
- m, err := c.Recv()
- if err != nil {
- return err
- }
- outMessage := anyMessage.(*sentryproto.InboundMessage)
- proto.Merge(outMessage, m)
- return nil
-}
-
-// -- end Messages
-// -- start Peers
-
func (c *SentryClientDirect) PeerEvents(ctx context.Context, in *sentryproto.PeerEventsRequest, opts ...grpc.CallOption) (sentryproto.Sentry_PeerEventsClient, error) {
- ch := make(chan *peersReply, libsentry.MessagesQueueSize)
- streamServer := &SentryPeersStreamS{ch: ch, ctx: ctx}
+ ch := make(chan libsentry.StreamReply[*sentryproto.PeerEvent], libsentry.MessagesQueueSize)
+ streamServer := &libsentry.SentryStreamS[*sentryproto.PeerEvent]{Ch: ch, Ctx: ctx}
go func() {
defer close(ch)
streamServer.Err(c.server.PeerEvents(in, streamServer))
}()
- return &SentryPeersStreamC{ch: ch, ctx: ctx}, nil
+ return &libsentry.SentryStreamC[*sentryproto.PeerEvent]{Ch: ch, Ctx: ctx}, nil
}
func (c *SentryClientDirect) AddPeer(ctx context.Context, in *sentryproto.AddPeerRequest, opts ...grpc.CallOption) (*sentryproto.AddPeerReply, error) {
@@ -340,61 +280,6 @@ func (c *SentryClientDirect) RemoveTrustedPeer(ctx context.Context, in *sentrypr
return c.server.RemoveTrustedPeer(ctx, in)
}
-type peersReply struct {
- r *sentryproto.PeerEvent
- err error
-}
-
-// SentryPeersStreamS - implements proto_sentryproto.Sentry_ReceivePeersServer
-type SentryPeersStreamS struct {
- ch chan *peersReply
- ctx context.Context
- grpc.ServerStream
-}
-
-func (s *SentryPeersStreamS) Send(m *sentryproto.PeerEvent) error {
- s.ch <- &peersReply{r: m}
- libsentry.EvictOldestIfHalfFull(s.ch)
- return nil
-}
-
-func (s *SentryPeersStreamS) Context() context.Context { return s.ctx }
-
-func (s *SentryPeersStreamS) Err(err error) {
- if err == nil {
- return
- }
- s.ch <- &peersReply{err: err}
-}
-
-type SentryPeersStreamC struct {
- ch chan *peersReply
- ctx context.Context
- grpc.ClientStream
-}
-
-func (c *SentryPeersStreamC) Recv() (*sentryproto.PeerEvent, error) {
- m, ok := <-c.ch
- if !ok || m == nil {
- return nil, io.EOF
- }
- return m.r, m.err
-}
-
-func (c *SentryPeersStreamC) Context() context.Context { return c.ctx }
-
-func (c *SentryPeersStreamC) RecvMsg(anyMessage any) error {
- m, err := c.Recv()
- if err != nil {
- return err
- }
- outMessage := anyMessage.(*sentryproto.PeerEvent)
- proto.Merge(outMessage, m)
- return nil
-}
-
-// -- end Peers
-
func (c *SentryClientDirect) NodeInfo(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*typesproto.NodeInfoReply, error) {
return c.server.NodeInfo(ctx, in)
}
diff --git a/p2p/sentry/libsentry/protocol.go b/p2p/sentry/libsentry/protocol.go
index ef603b6565b..f978df71d8a 100644
--- a/p2p/sentry/libsentry/protocol.go
+++ b/p2p/sentry/libsentry/protocol.go
@@ -20,6 +20,8 @@ import (
"github.com/erigontech/erigon/node/gointerfaces/sentryproto"
)
+const noProtocol sentryproto.Protocol = -1
+
// ethProtocolsByVersion lists ETH protocols in ascending version order.
// Used by MinProtocol to find the lowest version supporting a message.
var ethProtocolsByVersion = []sentryproto.Protocol{
@@ -38,7 +40,7 @@ func MinProtocol(m sentryproto.MessageId) sentryproto.Protocol {
}
}
- return -1
+ return noProtocol
}
var ProtoIds = map[sentryproto.Protocol]map[sentryproto.MessageId]struct{}{
diff --git a/p2p/sentry/libsentry/sentrymultiplexer.go b/p2p/sentry/libsentry/sentrymultiplexer.go
index c9329b53fc4..02bf3b496a6 100644
--- a/p2p/sentry/libsentry/sentrymultiplexer.go
+++ b/p2p/sentry/libsentry/sentrymultiplexer.go
@@ -29,7 +29,6 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/reflect/protoreflect"
"google.golang.org/protobuf/types/known/emptypb"
@@ -55,136 +54,178 @@ func NewSentryMultiplexer(clients []sentryproto.SentryClient) *sentryMultiplexer
mux := &sentryMultiplexer{}
mux.clients = make([]*client, len(clients))
for i, c := range clients {
- mux.clients[i] = &client{sync.RWMutex{}, c, -1}
+ mux.clients[i] = &client{sync.RWMutex{}, c, noProtocol}
}
return mux
}
-func (m *sentryMultiplexer) SetStatus(ctx context.Context, in *sentryproto.StatusData, opts ...grpc.CallOption) (*sentryproto.SetStatusReply, error) {
+// fanOut calls call concurrently on every client whose negotiated protocol is
+// at least minProtocol (noProtocol admits clients that have not handshaken
+// yet) and feeds each reply to collect (may be nil) under a shared mutex. A
+// non-nil error from call or collect cancels the remaining calls.
+func fanOut[R any](ctx context.Context, clients []*client, minProtocol sentryproto.Protocol, call func(context.Context, *client) (R, error), collect func(clientIndex int, reply R) error) error {
g, gctx := errgroup.WithContext(ctx)
+ var mu sync.Mutex
- for _, client := range m.clients {
+ for i, client := range clients {
- client.RLock()
- protocol := client.protocol
- client.RUnlock()
+ if minProtocol > noProtocol {
+ client.RLock()
+ protocol := client.protocol
+ client.RUnlock()
- if protocol >= 0 {
- g.Go(func() error {
- _, err := client.SetStatus(gctx, in, opts...)
- return err
- })
+ if protocol < minProtocol {
+ continue
+ }
}
- }
-
- err := g.Wait()
- if err != nil {
- return nil, err
- }
+ g.Go(func() error {
+ reply, err := call(gctx, client)
- return &sentryproto.SetStatusReply{}, nil
-}
+ if err != nil {
+ return err
+ }
-func (m *sentryMultiplexer) PenalizePeer(ctx context.Context, in *sentryproto.PenalizePeerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
- g, gctx := errgroup.WithContext(ctx)
+ if collect == nil {
+ return nil
+ }
- for _, client := range m.clients {
+ mu.Lock()
+ defer mu.Unlock()
- g.Go(func() error {
- _, err := client.PenalizePeer(gctx, in, opts...)
- return err
+ return collect(i, reply)
})
}
- return &emptypb.Empty{}, g.Wait()
+ return g.Wait()
}
-func (m *sentryMultiplexer) SetPeerLatestBlock(ctx context.Context, in *sentryproto.SetPeerLatestBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- for _, client := range m.clients {
+func fanOutSuccess[R interface{ GetSuccess() bool }](ctx context.Context, clients []*client, call func(context.Context, *client) (R, error)) (bool, error) {
+ var success bool
- g.Go(func() error {
- _, err := client.SetPeerLatestBlock(gctx, in, opts...)
- return err
- })
- }
+ err := fanOut(ctx, clients, noProtocol, call, func(_ int, reply R) error {
+ success = success || reply.GetSuccess()
+ return nil
+ })
- return &emptypb.Empty{}, g.Wait()
+ return success, err
}
-func (m *sentryMultiplexer) SetPeerMinimumBlock(ctx context.Context, in *sentryproto.SetPeerMinimumBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+func streamFanIn[T protoreflect.ProtoMessage, S interface{ Recv() (T, error) }](ctx context.Context, clients []*client, open func(context.Context, *client) (S, error)) *SentryStreamC[T] {
g, gctx := errgroup.WithContext(ctx)
- for _, client := range m.clients {
+ ch := make(chan StreamReply[T], MessagesQueueSize)
+ streamServer := &SentryStreamS[T]{Ch: ch, Ctx: ctx}
- g.Go(func() error {
- _, err := client.SetPeerMinimumBlock(gctx, in, opts...)
- return err
- })
+ go func() {
+ defer close(ch)
+
+ for _, client := range clients {
+
+ g.Go(func() error {
+ stream, err := open(gctx, client)
+
+ if err != nil {
+ streamServer.Err(err)
+ return err
+ }
+
+ for {
+ message, err := stream.Recv()
+
+ if err != nil {
+ if errors.Is(err, io.EOF) {
+ return nil
+ }
+
+ streamServer.Err(err)
+
+ select {
+ case <-gctx.Done():
+ return gctx.Err()
+ default:
+ }
+
+ return fmt.Errorf("recv: %w", err)
+ }
+
+ streamServer.Send(message)
+ }
+ })
+ }
+
+ g.Wait()
+ }()
+
+ return &SentryStreamC[T]{Ch: ch, Ctx: ctx}
+}
+
+func (m *sentryMultiplexer) SetStatus(ctx context.Context, in *sentryproto.StatusData, opts ...grpc.CallOption) (*sentryproto.SetStatusReply, error) {
+ err := fanOut(ctx, m.clients, 0, func(gctx context.Context, client *client) (*sentryproto.SetStatusReply, error) {
+ return client.SetStatus(gctx, in, opts...)
+ }, nil)
+
+ if err != nil {
+ return nil, err
}
- return &emptypb.Empty{}, g.Wait()
+ return &sentryproto.SetStatusReply{}, nil
}
-func (m *sentryMultiplexer) SetPeerBlockRange(ctx context.Context, in *sentryproto.SetPeerBlockRangeRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
- g, gctx := errgroup.WithContext(ctx)
+func (m *sentryMultiplexer) PenalizePeer(ctx context.Context, in *sentryproto.PenalizePeerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+ return &emptypb.Empty{}, fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*emptypb.Empty, error) {
+ return client.PenalizePeer(gctx, in, opts...)
+ }, nil)
+}
- for _, client := range m.clients {
+func (m *sentryMultiplexer) SetPeerLatestBlock(ctx context.Context, in *sentryproto.SetPeerLatestBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+ return &emptypb.Empty{}, fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*emptypb.Empty, error) {
+ return client.SetPeerLatestBlock(gctx, in, opts...)
+ }, nil)
+}
- g.Go(func() error {
- _, err := client.SetPeerBlockRange(gctx, in, opts...)
- return err
- })
- }
+func (m *sentryMultiplexer) SetPeerMinimumBlock(ctx context.Context, in *sentryproto.SetPeerMinimumBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+ return &emptypb.Empty{}, fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*emptypb.Empty, error) {
+ return client.SetPeerMinimumBlock(gctx, in, opts...)
+ }, nil)
+}
- return &emptypb.Empty{}, g.Wait()
+func (m *sentryMultiplexer) SetPeerBlockRange(ctx context.Context, in *sentryproto.SetPeerBlockRangeRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+ return &emptypb.Empty{}, fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*emptypb.Empty, error) {
+ return client.SetPeerBlockRange(gctx, in, opts...)
+ }, nil)
}
// HandShake is not performed on the multi-client level
func (m *sentryMultiplexer) HandShake(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*sentryproto.HandShakeReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
var protocol sentryproto.Protocol
- var mu sync.Mutex
-
- for _, client := range m.clients {
+ err := fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (sentryproto.Protocol, error) {
client.RLock()
clientProtocol := client.protocol
client.RUnlock()
if clientProtocol >= 0 {
- mu.Lock()
- if clientProtocol > protocol {
- protocol = clientProtocol
- }
- mu.Unlock()
- continue
+ return clientProtocol, nil
}
- g.Go(func() error {
- reply, err := client.HandShake(gctx, &emptypb.Empty{}, grpc.WaitForReady(true))
- if err != nil {
- return err
- }
-
- mu.Lock()
- if reply.Protocol > protocol {
- protocol = reply.Protocol
- }
- mu.Unlock()
+ reply, err := client.HandShake(gctx, &emptypb.Empty{}, grpc.WaitForReady(true))
- client.Lock()
- client.protocol = reply.Protocol
- client.Unlock()
+ if err != nil {
+ return noProtocol, err
+ }
- return nil
- })
- }
+ client.Lock()
+ client.protocol = reply.Protocol
+ client.Unlock()
- err := g.Wait()
+ return reply.Protocol, nil
+ }, func(_ int, clientProtocol sentryproto.Protocol) error {
+ if clientProtocol > protocol {
+ protocol = clientProtocol
+ }
+ return nil
+ })
if err != nil {
return nil, err
@@ -439,150 +480,22 @@ func (m *sentryMultiplexer) SendMessageToAll(ctx context.Context, in *sentryprot
return &sentryproto.SentPeers{Peers: allSentPeers}, nil
}
-type StreamReply[T protoreflect.ProtoMessage] struct {
- R T
- Err error
-}
-
-// SentryMessagesStreamS implements proto_sentry.Sentry_ReceiveMessagesServer
-type SentryStreamS[T protoreflect.ProtoMessage] struct {
- Ch chan StreamReply[T]
- Ctx context.Context
- grpc.ServerStream
-}
-
-func (s *SentryStreamS[T]) Send(m T) error {
- s.Ch <- StreamReply[T]{R: m}
- EvictOldestIfHalfFull(s.Ch)
- return nil
-}
-
-func (s *SentryStreamS[T]) Context() context.Context { return s.Ctx }
-
-func (s *SentryStreamS[T]) Err(err error) {
- if err == nil {
- return
- }
- s.Ch <- StreamReply[T]{Err: err}
-}
-
-func (s *SentryStreamS[T]) Close() {
- if s.Ch != nil {
- ch := s.Ch
- s.Ch = nil
- close(ch)
- }
-}
-
-type SentryStreamC[T protoreflect.ProtoMessage] struct {
- Ch chan StreamReply[T]
- Ctx context.Context
- grpc.ClientStream
-}
-
-func (c *SentryStreamC[T]) Recv() (T, error) {
- m, ok := <-c.Ch
- if !ok {
- var t T
- return t, io.EOF
- }
- return m.R, m.Err
-}
-
-func (c *SentryStreamC[T]) Context() context.Context { return c.Ctx }
-
-func (c *SentryStreamC[T]) RecvMsg(anyMessage any) error {
- m, err := c.Recv()
- if err != nil {
- return err
- }
- outMessage := anyMessage.(T)
- proto.Merge(outMessage, m)
- return nil
-}
-
func (m *sentryMultiplexer) Messages(ctx context.Context, in *sentryproto.MessagesRequest, opts ...grpc.CallOption) (sentryproto.Sentry_MessagesClient, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- ch := make(chan StreamReply[*sentryproto.InboundMessage], MessagesQueueSize)
- streamServer := &SentryStreamS[*sentryproto.InboundMessage]{Ch: ch, Ctx: ctx}
-
- go func() {
- defer close(ch)
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- messages, err := client.Messages(gctx, in, opts...)
-
- if err != nil {
- streamServer.Err(err)
- return err
- }
-
- for {
- inboundMessage, err := messages.Recv()
-
- if err != nil {
- if errors.Is(err, io.EOF) {
- return nil
- }
-
- streamServer.Err(err)
-
- select {
- case <-gctx.Done():
- return gctx.Err()
- default:
- }
-
- return fmt.Errorf("recv: %w", err)
- }
-
- streamServer.Send(inboundMessage)
- }
- })
- }
-
- g.Wait()
- }()
-
- return &SentryStreamC[*sentryproto.InboundMessage]{Ch: ch, Ctx: ctx}, nil
+ return streamFanIn(ctx, m.clients,
+ func(gctx context.Context, client *client) (sentryproto.Sentry_MessagesClient, error) {
+ return client.Messages(gctx, in, opts...)
+ }), nil
}
func (m *sentryMultiplexer) peersByClient(ctx context.Context, minProtocol sentryproto.Protocol, opts ...grpc.CallOption) ([]*sentryproto.PeersReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- var allReplies = make([]*sentryproto.PeersReply, len(m.clients))
- var allMutex sync.RWMutex
+ allReplies := make([]*sentryproto.PeersReply, len(m.clients))
- for i, client := range m.clients {
-
- client.RLock()
- protocol := client.protocol
- client.RUnlock()
-
- if protocol < minProtocol {
- continue
- }
-
- g.Go(func() error {
- sentPeers, err := client.Peers(gctx, &emptypb.Empty{}, opts...)
-
- if err != nil {
- return err
- }
-
- allMutex.Lock()
- defer allMutex.Unlock()
-
- allReplies[i] = sentPeers
-
- return nil
- })
- }
-
- err := g.Wait()
+ err := fanOut(ctx, m.clients, minProtocol, func(gctx context.Context, client *client) (*sentryproto.PeersReply, error) {
+ return client.Peers(gctx, &emptypb.Empty{}, opts...)
+ }, func(clientIndex int, reply *sentryproto.PeersReply) error {
+ allReplies[clientIndex] = reply
+ return nil
+ })
if err != nil {
return nil, err
@@ -594,7 +507,7 @@ func (m *sentryMultiplexer) peersByClient(ctx context.Context, minProtocol sentr
func (m *sentryMultiplexer) Peers(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*sentryproto.PeersReply, error) {
var allPeers []*typesproto.PeerInfo
- allReplies, err := m.peersByClient(ctx, -1, opts...)
+ allReplies, err := m.peersByClient(ctx, noProtocol, opts...)
if err != nil {
return nil, err
@@ -608,30 +521,14 @@ func (m *sentryMultiplexer) Peers(ctx context.Context, in *emptypb.Empty, opts .
}
func (m *sentryMultiplexer) PeerCount(ctx context.Context, in *sentryproto.PeerCountRequest, opts ...grpc.CallOption) (*sentryproto.PeerCountReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
var allCount uint64
- var allMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- peerCount, err := client.PeerCount(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- allMutex.Lock()
- defer allMutex.Unlock()
-
- allCount += peerCount.GetCount()
-
- return nil
- })
- }
- err := g.Wait()
+ err := fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*sentryproto.PeerCountReply, error) {
+ return client.PeerCount(gctx, in, opts...)
+ }, func(_ int, reply *sentryproto.PeerCountReply) error {
+ allCount += reply.GetCount()
+ return nil
+ })
if err != nil {
return nil, err
@@ -643,37 +540,21 @@ func (m *sentryMultiplexer) PeerCount(ctx context.Context, in *sentryproto.PeerC
var errFound = fmt.Errorf("found peer")
func (m *sentryMultiplexer) PeerById(ctx context.Context, in *sentryproto.PeerByIdRequest, opts ...grpc.CallOption) (*sentryproto.PeerByIdReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
var peer *typesproto.PeerInfo
- var peerMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- reply, err := client.PeerById(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- peerMutex.Lock()
- defer peerMutex.Unlock()
-
- if peer == nil && reply.GetPeer() != nil {
- peer = reply.GetPeer()
- // return a success error here to have the
- // group stop other concurrent requests
- return errFound
- }
- return nil
- })
- }
-
- err := g.Wait()
+ err := fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*sentryproto.PeerByIdReply, error) {
+ return client.PeerById(gctx, in, opts...)
+ }, func(_ int, reply *sentryproto.PeerByIdReply) error {
+ if peer == nil && reply.GetPeer() != nil {
+ peer = reply.GetPeer()
+ // return a success error here to have the
+ // group stop other concurrent requests
+ return errFound
+ }
+ return nil
+ })
- if err != nil && !errors.Is(errFound, err) {
+ if err != nil && !errors.Is(err, errFound) {
return nil, err
}
@@ -681,82 +562,16 @@ func (m *sentryMultiplexer) PeerById(ctx context.Context, in *sentryproto.PeerBy
}
func (m *sentryMultiplexer) PeerEvents(ctx context.Context, in *sentryproto.PeerEventsRequest, opts ...grpc.CallOption) (sentryproto.Sentry_PeerEventsClient, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- ch := make(chan StreamReply[*sentryproto.PeerEvent], MessagesQueueSize)
- streamServer := &SentryStreamS[*sentryproto.PeerEvent]{Ch: ch, Ctx: ctx}
-
- go func() {
- defer close(ch)
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- messages, err := client.PeerEvents(gctx, in, opts...)
-
- if err != nil {
- streamServer.Err(err)
- return err
- }
-
- for {
- inboundMessage, err := messages.Recv()
-
- if err != nil {
- if errors.Is(err, io.EOF) {
- return nil
- }
-
- streamServer.Err(err)
-
- select {
- case <-gctx.Done():
- return gctx.Err()
- default:
- }
-
- return fmt.Errorf("recv: %w", err)
- }
-
- streamServer.Send(inboundMessage)
- }
- })
- }
-
- g.Wait()
- }()
-
- return &SentryStreamC[*sentryproto.PeerEvent]{Ch: ch, Ctx: ctx}, nil
+ return streamFanIn(ctx, m.clients,
+ func(gctx context.Context, client *client) (sentryproto.Sentry_PeerEventsClient, error) {
+ return client.PeerEvents(gctx, in, opts...)
+ }), nil
}
func (m *sentryMultiplexer) AddPeer(ctx context.Context, in *sentryproto.AddPeerRequest, opts ...grpc.CallOption) (*sentryproto.AddPeerReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- var success bool
- var successMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- result, err := client.AddPeer(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- successMutex.Lock()
- defer successMutex.Unlock()
-
- // if any client returns success return success
- if !success && result.GetSuccess() {
- success = true
- }
-
- return nil
- })
- }
-
- err := g.Wait()
+ success, err := fanOutSuccess(ctx, m.clients, func(gctx context.Context, client *client) (*sentryproto.AddPeerReply, error) {
+ return client.AddPeer(gctx, in, opts...)
+ })
if err != nil {
return nil, err
@@ -766,33 +581,9 @@ func (m *sentryMultiplexer) AddPeer(ctx context.Context, in *sentryproto.AddPeer
}
func (m *sentryMultiplexer) RemovePeer(ctx context.Context, in *sentryproto.RemovePeerRequest, opts ...grpc.CallOption) (*sentryproto.RemovePeerReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- var success bool
- var successMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- result, err := client.RemovePeer(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- successMutex.Lock()
- defer successMutex.Unlock()
-
- // if any client returns success return success
- if !success && result.GetSuccess() {
- success = true
- }
-
- return nil
- })
- }
-
- err := g.Wait()
+ success, err := fanOutSuccess(ctx, m.clients, func(gctx context.Context, client *client) (*sentryproto.RemovePeerReply, error) {
+ return client.RemovePeer(gctx, in, opts...)
+ })
if err != nil {
return nil, err
@@ -802,33 +593,9 @@ func (m *sentryMultiplexer) RemovePeer(ctx context.Context, in *sentryproto.Remo
}
func (m *sentryMultiplexer) AddTrustedPeer(ctx context.Context, in *sentryproto.AddPeerRequest, opts ...grpc.CallOption) (*sentryproto.AddPeerReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- var success bool
- var successMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- result, err := client.AddTrustedPeer(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- successMutex.Lock()
- defer successMutex.Unlock()
-
- // if any client returns success return success
- if !success && result.GetSuccess() {
- success = true
- }
-
- return nil
- })
- }
-
- err := g.Wait()
+ success, err := fanOutSuccess(ctx, m.clients, func(gctx context.Context, client *client) (*sentryproto.AddPeerReply, error) {
+ return client.AddTrustedPeer(gctx, in, opts...)
+ })
if err != nil {
return nil, err
@@ -838,33 +605,9 @@ func (m *sentryMultiplexer) AddTrustedPeer(ctx context.Context, in *sentryproto.
}
func (m *sentryMultiplexer) RemoveTrustedPeer(ctx context.Context, in *sentryproto.RemovePeerRequest, opts ...grpc.CallOption) (*sentryproto.RemovePeerReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
- var success bool
- var successMutex sync.RWMutex
-
- for _, client := range m.clients {
-
- g.Go(func() error {
- result, err := client.RemoveTrustedPeer(gctx, in, opts...)
-
- if err != nil {
- return err
- }
-
- successMutex.Lock()
- defer successMutex.Unlock()
-
- // if any client returns success return success
- if !success && result.GetSuccess() {
- success = true
- }
-
- return nil
- })
- }
-
- err := g.Wait()
+ success, err := fanOutSuccess(ctx, m.clients, func(gctx context.Context, client *client) (*sentryproto.RemovePeerReply, error) {
+ return client.RemoveTrustedPeer(gctx, in, opts...)
+ })
if err != nil {
return nil, err
@@ -878,30 +621,14 @@ func (m *sentryMultiplexer) NodeInfo(ctx context.Context, in *emptypb.Empty, opt
}
func (m *sentryMultiplexer) NodeInfos(ctx context.Context, opts ...grpc.CallOption) ([]*typesproto.NodeInfoReply, error) {
- g, gctx := errgroup.WithContext(ctx)
-
var allInfos []*typesproto.NodeInfoReply
- var allMutex sync.RWMutex
-
- for _, client := range m.clients {
- g.Go(func() error {
- info, err := client.NodeInfo(gctx, &emptypb.Empty{}, opts...)
-
- if err != nil {
- return err
- }
-
- allMutex.Lock()
- defer allMutex.Unlock()
-
- allInfos = append(allInfos, info)
-
- return nil
- })
- }
-
- err := g.Wait()
+ err := fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (*typesproto.NodeInfoReply, error) {
+ return client.NodeInfo(gctx, &emptypb.Empty{}, opts...)
+ }, func(_ int, info *typesproto.NodeInfoReply) error {
+ allInfos = append(allInfos, info)
+ return nil
+ })
if err != nil {
return nil, err
diff --git a/p2p/sentry/libsentry/stream.go b/p2p/sentry/libsentry/stream.go
new file mode 100644
index 00000000000..5d3dd9b4317
--- /dev/null
+++ b/p2p/sentry/libsentry/stream.go
@@ -0,0 +1,87 @@
+// Copyright 2024 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package libsentry
+
+import (
+ "context"
+ "io"
+
+ "google.golang.org/grpc"
+ "google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/reflect/protoreflect"
+)
+
+type StreamReply[T protoreflect.ProtoMessage] struct {
+ R T
+ Err error
+}
+
+type SentryStreamS[T protoreflect.ProtoMessage] struct {
+ Ch chan StreamReply[T]
+ Ctx context.Context
+ grpc.ServerStream
+}
+
+func (s *SentryStreamS[T]) Send(m T) error {
+ s.Ch <- StreamReply[T]{R: m}
+ EvictOldestIfHalfFull(s.Ch)
+ return nil
+}
+
+func (s *SentryStreamS[T]) Context() context.Context { return s.Ctx }
+
+func (s *SentryStreamS[T]) Err(err error) {
+ if err == nil {
+ return
+ }
+ s.Ch <- StreamReply[T]{Err: err}
+}
+
+func (s *SentryStreamS[T]) Close() {
+ if s.Ch != nil {
+ ch := s.Ch
+ s.Ch = nil
+ close(ch)
+ }
+}
+
+type SentryStreamC[T protoreflect.ProtoMessage] struct {
+ Ch chan StreamReply[T]
+ Ctx context.Context
+ grpc.ClientStream
+}
+
+func (c *SentryStreamC[T]) Recv() (T, error) {
+ m, ok := <-c.Ch
+ if !ok {
+ var t T
+ return t, io.EOF
+ }
+ return m.R, m.Err
+}
+
+func (c *SentryStreamC[T]) Context() context.Context { return c.Ctx }
+
+func (c *SentryStreamC[T]) RecvMsg(anyMessage any) error {
+ m, err := c.Recv()
+ if err != nil {
+ return err
+ }
+ outMessage := anyMessage.(T)
+ proto.Merge(outMessage, m)
+ return nil
+}
diff --git a/node/direct/sentry_client_test.go b/p2p/sentry/libsentry/stream_test.go
similarity index 71%
rename from node/direct/sentry_client_test.go
rename to p2p/sentry/libsentry/stream_test.go
index c8ec90e9354..17370bf370f 100644
--- a/node/direct/sentry_client_test.go
+++ b/p2p/sentry/libsentry/stream_test.go
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License
// along with Erigon. If not, see .
-package direct
+package libsentry_test
import (
"encoding/binary"
@@ -31,9 +31,9 @@ import (
// (the pre-fix 16384 chan would hold ~160 GB of peer-controlled payload at
// the eth 10 MB limit). Eviction must drop oldest entries, preserving the
// freshest.
-func TestSentryMessagesStreamS_SendEvictsWhenConsumerSlow(t *testing.T) {
- ch := make(chan *inboundMessageReply, libsentry.MessagesQueueSize)
- s := &SentryMessagesStreamS{ch: ch, ctx: t.Context()}
+func TestSentryStreamS_SendEvictsWhenConsumerSlow(t *testing.T) {
+ ch := make(chan libsentry.StreamReply[*sentryproto.InboundMessage], libsentry.MessagesQueueSize)
+ s := &libsentry.SentryStreamS[*sentryproto.InboundMessage]{Ch: ch, Ctx: t.Context()}
const flood = libsentry.MessagesQueueSize * 10
for i := range flood {
@@ -48,23 +48,9 @@ func TestSentryMessagesStreamS_SendEvictsWhenConsumerSlow(t *testing.T) {
var last uint32
for len(ch) > 0 {
r := <-ch
- require.NotNil(t, r)
- require.NotNil(t, r.r)
- last = binary.LittleEndian.Uint32(r.r.Data)
+ require.NotNil(t, r.R)
+ last = binary.LittleEndian.Uint32(r.R.Data)
}
assert.Equal(t, uint32(flood-1), last,
"eviction drops from the front; the most recent Send must remain queued")
}
-
-func TestSentryPeersStreamS_SendEvictsWhenConsumerSlow(t *testing.T) {
- ch := make(chan *peersReply, libsentry.MessagesQueueSize)
- s := &SentryPeersStreamS{ch: ch, ctx: t.Context()}
-
- const flood = libsentry.MessagesQueueSize * 10
- for range flood {
- require.NoError(t, s.Send(&sentryproto.PeerEvent{}))
- }
-
- require.LessOrEqual(t, len(ch), cap(ch))
- require.Less(t, len(ch), flood)
-}
diff --git a/p2p/sentry/sentrymultiplexer_test.go b/p2p/sentry/sentrymultiplexer_test.go
index ec2d61948fd..dcd6fb45380 100644
--- a/p2p/sentry/sentrymultiplexer_test.go
+++ b/p2p/sentry/sentrymultiplexer_test.go
@@ -5,6 +5,7 @@ import (
"crypto/ecdsa"
"crypto/rand"
"encoding/hex"
+ "errors"
"fmt"
"io"
"net"
@@ -228,6 +229,104 @@ func TestSend(t *testing.T) {
require.Equal(t, 10, statusCount)
}
+func TestPeerAdminAnySuccess(t *testing.T) {
+ testErr := errors.New("test error")
+
+ methods := []struct {
+ name string
+ expect func(client *direct.MockSentryClient, success bool, err error)
+ call func(mux sentryproto.SentryClient) (bool, error)
+ }{
+ {
+ name: "AddPeer",
+ expect: func(client *direct.MockSentryClient, success bool, err error) {
+ client.EXPECT().AddPeer(gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(&sentryproto.AddPeerReply{Success: success}, err).AnyTimes()
+ },
+ call: func(mux sentryproto.SentryClient) (bool, error) {
+ reply, err := mux.AddPeer(context.Background(), &sentryproto.AddPeerRequest{})
+ return reply.GetSuccess(), err
+ },
+ },
+ {
+ name: "RemovePeer",
+ expect: func(client *direct.MockSentryClient, success bool, err error) {
+ client.EXPECT().RemovePeer(gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(&sentryproto.RemovePeerReply{Success: success}, err).AnyTimes()
+ },
+ call: func(mux sentryproto.SentryClient) (bool, error) {
+ reply, err := mux.RemovePeer(context.Background(), &sentryproto.RemovePeerRequest{})
+ return reply.GetSuccess(), err
+ },
+ },
+ {
+ name: "AddTrustedPeer",
+ expect: func(client *direct.MockSentryClient, success bool, err error) {
+ client.EXPECT().AddTrustedPeer(gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(&sentryproto.AddPeerReply{Success: success}, err).AnyTimes()
+ },
+ call: func(mux sentryproto.SentryClient) (bool, error) {
+ reply, err := mux.AddTrustedPeer(context.Background(), &sentryproto.AddPeerRequest{})
+ return reply.GetSuccess(), err
+ },
+ },
+ {
+ name: "RemoveTrustedPeer",
+ expect: func(client *direct.MockSentryClient, success bool, err error) {
+ client.EXPECT().RemoveTrustedPeer(gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(&sentryproto.RemovePeerReply{Success: success}, err).AnyTimes()
+ },
+ call: func(mux sentryproto.SentryClient) (bool, error) {
+ reply, err := mux.RemoveTrustedPeer(context.Background(), &sentryproto.RemovePeerRequest{})
+ return reply.GetSuccess(), err
+ },
+ },
+ }
+
+ scenarios := []struct {
+ name string
+ successes []bool
+ errs []error
+ wantSuccess bool
+ }{
+ {name: "all clients fail", successes: []bool{false, false, false}},
+ {name: "one client succeeds", successes: []bool{false, true, false}, wantSuccess: true},
+ {name: "all clients succeed", successes: []bool{true, true, true}, wantSuccess: true},
+ {name: "client error", successes: []bool{true, true, true}, errs: []error{nil, testErr, nil}},
+ }
+
+ for _, method := range methods {
+ for _, scenario := range scenarios {
+ t.Run(method.name+"/"+scenario.name, func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ clients := make([]sentryproto.SentryClient, 0, len(scenario.successes))
+ for i, success := range scenario.successes {
+ client := newClient(ctrl, i, nil)
+ var err error
+ if scenario.errs != nil {
+ err = scenario.errs[i]
+ }
+ method.expect(client, success, err)
+ clients = append(clients, client)
+ }
+
+ mux := libsentry.NewSentryMultiplexer(clients)
+ success, err := method.call(mux)
+
+ if scenario.errs != nil {
+ require.ErrorIs(t, err, testErr)
+ return
+ }
+
+ require.NoError(t, err)
+ require.Equal(t, scenario.wantSuccess, success)
+ })
+ }
+ }
+}
+
func TestMessages(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()