From b60428657b947ceec6d8de580e9352c505b0a3e7 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 21:15:02 +0200 Subject: [PATCH 1/4] p2p/sentry: pin any-client-success semantics of multiplexer peer admin methods Characterization test ahead of a multiplexer fan-out refactor: AddPeer, RemovePeer, AddTrustedPeer and RemoveTrustedPeer must report success when any underlying client succeeds, and propagate the first client error. Only AddPeer had (indirect) coverage before. --- p2p/sentry/sentrymultiplexer_test.go | 99 ++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) 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() From 7a424eedc99f5f6325a1fd967138bac121b00d2c Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 21:17:10 +0200 Subject: [PATCH 2/4] p2p/sentry/libsentry, node/direct: dedupe sentry stream adapters Move the generic SentryStreamS/SentryStreamC/StreamReply adapters from sentrymultiplexer.go into their own libsentry/stream.go, and use them from SentryClientDirect.Messages/PeerEvents instead of the hand-monomorphized SentryMessagesStreamS/C and SentryPeersStreamS/C copies (identical Send with EvictOldestIfHalfFull, Err, Recv, RecvMsg with proto.Merge), which are deleted. The exported copies were referenced only by their own tests. The two node/direct eviction tests pinned one monomorphic copy each; with a single generic implementation left, they collapse into one test moved to libsentry/stream_test.go, keeping the stronger drop-oldest assertions. --- node/direct/sentry_client.go | 127 +----------------- p2p/sentry/libsentry/sentrymultiplexer.go | 64 --------- p2p/sentry/libsentry/stream.go | 87 ++++++++++++ .../sentry/libsentry/stream_test.go | 26 +--- 4 files changed, 99 insertions(+), 205 deletions(-) create mode 100644 p2p/sentry/libsentry/stream.go rename node/direct/sentry_client_test.go => p2p/sentry/libsentry/stream_test.go (71%) 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/sentrymultiplexer.go b/p2p/sentry/libsentry/sentrymultiplexer.go index c9329b53fc4..73943d7ba3f 100644 --- a/p2p/sentry/libsentry/sentrymultiplexer.go +++ b/p2p/sentry/libsentry/sentrymultiplexer.go @@ -29,8 +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" "github.com/erigontech/erigon/common" @@ -439,68 +437,6 @@ 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) 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) -} From 0b21d44f0524c82ea695df4603898c4257ab1c47 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 21:21:42 +0200 Subject: [PATCH 3/4] p2p/sentry/libsentry: collapse multiplexer fan-out boilerplate with generics Every unary sentryMultiplexer method open-coded the same errgroup scaffold: spawn a goroutine per client, optionally skip clients below a protocol threshold, collect replies under a mutex, wait. Replace ~16 copies with - fanOut: concurrent per-client call with protocol gating (-1 admits un-handshaken clients, 0 reproduces the SetStatus negotiated-only filter) and mutex-guarded collect that can return an error to cancel the group; - fanOutSuccess: reply.GetSuccess() OR-reduction for the four peer admin methods; - streamFanIn: the Messages/PeerEvents stream fan-in, which was duplicated verbatim modulo the message type. Deliberate variants are preserved: PeerById still early-stops via the errFound sentinel, HandShake still skips already-handshaken clients while folding their cached protocol into the max and stores each fresh reply on the client, and SendMessageByMinBlock stays intentionally serial. Pure refactor pinned by the existing multiplexer tests plus the any-client-success characterization test added beforehand. --- p2p/sentry/libsentry/sentrymultiplexer.go | 547 +++++++--------------- 1 file changed, 168 insertions(+), 379 deletions(-) diff --git a/p2p/sentry/libsentry/sentrymultiplexer.go b/p2p/sentry/libsentry/sentrymultiplexer.go index 73943d7ba3f..7f3fedd0f2a 100644 --- a/p2p/sentry/libsentry/sentrymultiplexer.go +++ b/p2p/sentry/libsentry/sentrymultiplexer.go @@ -29,6 +29,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/reflect/protoreflect" "google.golang.org/protobuf/types/known/emptypb" "github.com/erigontech/erigon/common" @@ -58,131 +59,171 @@ func NewSentryMultiplexer(clients []sentryproto.SentryClient) *sentryMultiplexer 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 (-1 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 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, -1, 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, -1, 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, -1, 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, -1, 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, -1, 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, -1, 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 -1, 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 @@ -438,87 +479,21 @@ func (m *sentryMultiplexer) SendMessageToAll(ctx context.Context, in *sentryprot } 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[*sentryproto.InboundMessage](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 - - 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...) + allReplies := make([]*sentryproto.PeersReply, len(m.clients)) - 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 @@ -544,30 +519,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, -1, 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 @@ -579,35 +538,19 @@ 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, -1, 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) { return nil, err @@ -617,82 +560,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[*sentryproto.PeerEvent](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 @@ -702,33 +579,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 @@ -738,33 +591,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 @@ -774,33 +603,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 @@ -814,30 +619,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, -1, 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 From 2a0bf1c07c1b21b78c79ac8ef296fdf1c31d06f3 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:18:58 +0200 Subject: [PATCH 4/4] p2p/sentry/libsentry: name the noProtocol sentinel, skip the no-op fanOut gate Review follow-ups for the generics refactor: - Name the -1 "no protocol" sentinel (noProtocol) and use it for the client initializer, the admit-everyone minProtocol call sites, and MinProtocol's unknown-message return. - Skip fanOut's gating read when minProtocol is noProtocol: the gate cannot exclude anyone (client.protocol >= noProtocol always), the RLock was new overhead the pre-refactor ungated loops never had, and HandShake no longer reads client.protocol twice per client. - Idiomatic errors.Is argument order in PeerById. - Drop inferable streamFanIn type arguments (infertypeargs). --- p2p/sentry/libsentry/protocol.go | 4 +- p2p/sentry/libsentry/sentrymultiplexer.go | 48 ++++++++++++----------- 2 files changed, 28 insertions(+), 24 deletions(-) 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 7f3fedd0f2a..02bf3b496a6 100644 --- a/p2p/sentry/libsentry/sentrymultiplexer.go +++ b/p2p/sentry/libsentry/sentrymultiplexer.go @@ -54,27 +54,29 @@ 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 } // fanOut calls call concurrently on every client whose negotiated protocol is -// at least minProtocol (-1 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. +// 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 i, client := range clients { - client.RLock() - protocol := client.protocol - client.RUnlock() + if minProtocol > noProtocol { + client.RLock() + protocol := client.protocol + client.RUnlock() - if protocol < minProtocol { - continue + if protocol < minProtocol { + continue + } } g.Go(func() error { @@ -101,7 +103,7 @@ func fanOut[R any](ctx context.Context, clients []*client, minProtocol sentrypro func fanOutSuccess[R interface{ GetSuccess() bool }](ctx context.Context, clients []*client, call func(context.Context, *client) (R, error)) (bool, error) { var success bool - err := fanOut(ctx, clients, -1, call, func(_ int, reply R) error { + err := fanOut(ctx, clients, noProtocol, call, func(_ int, reply R) error { success = success || reply.GetSuccess() return nil }) @@ -171,25 +173,25 @@ func (m *sentryMultiplexer) SetStatus(ctx context.Context, in *sentryproto.Statu } func (m *sentryMultiplexer) PenalizePeer(ctx context.Context, in *sentryproto.PenalizePeerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { - return &emptypb.Empty{}, fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*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) } func (m *sentryMultiplexer) SetPeerLatestBlock(ctx context.Context, in *sentryproto.SetPeerLatestBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { - return &emptypb.Empty{}, fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*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) } func (m *sentryMultiplexer) SetPeerMinimumBlock(ctx context.Context, in *sentryproto.SetPeerMinimumBlockRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { - return &emptypb.Empty{}, fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*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) } func (m *sentryMultiplexer) SetPeerBlockRange(ctx context.Context, in *sentryproto.SetPeerBlockRangeRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { - return &emptypb.Empty{}, fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*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) } @@ -198,7 +200,7 @@ func (m *sentryMultiplexer) SetPeerBlockRange(ctx context.Context, in *sentrypro func (m *sentryMultiplexer) HandShake(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*sentryproto.HandShakeReply, error) { var protocol sentryproto.Protocol - err := fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (sentryproto.Protocol, error) { + err := fanOut(ctx, m.clients, noProtocol, func(gctx context.Context, client *client) (sentryproto.Protocol, error) { client.RLock() clientProtocol := client.protocol client.RUnlock() @@ -210,7 +212,7 @@ func (m *sentryMultiplexer) HandShake(ctx context.Context, in *emptypb.Empty, op reply, err := client.HandShake(gctx, &emptypb.Empty{}, grpc.WaitForReady(true)) if err != nil { - return -1, err + return noProtocol, err } client.Lock() @@ -479,7 +481,7 @@ func (m *sentryMultiplexer) SendMessageToAll(ctx context.Context, in *sentryprot } func (m *sentryMultiplexer) Messages(ctx context.Context, in *sentryproto.MessagesRequest, opts ...grpc.CallOption) (sentryproto.Sentry_MessagesClient, error) { - return streamFanIn[*sentryproto.InboundMessage](ctx, m.clients, + return streamFanIn(ctx, m.clients, func(gctx context.Context, client *client) (sentryproto.Sentry_MessagesClient, error) { return client.Messages(gctx, in, opts...) }), nil @@ -505,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 @@ -521,7 +523,7 @@ 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) { var allCount uint64 - err := fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*sentryproto.PeerCountReply, error) { + 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() @@ -540,7 +542,7 @@ var errFound = fmt.Errorf("found peer") func (m *sentryMultiplexer) PeerById(ctx context.Context, in *sentryproto.PeerByIdRequest, opts ...grpc.CallOption) (*sentryproto.PeerByIdReply, error) { var peer *typesproto.PeerInfo - err := fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*sentryproto.PeerByIdReply, error) { + 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 { @@ -552,7 +554,7 @@ func (m *sentryMultiplexer) PeerById(ctx context.Context, in *sentryproto.PeerBy return nil }) - if err != nil && !errors.Is(errFound, err) { + if err != nil && !errors.Is(err, errFound) { return nil, err } @@ -560,7 +562,7 @@ 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) { - return streamFanIn[*sentryproto.PeerEvent](ctx, m.clients, + return streamFanIn(ctx, m.clients, func(gctx context.Context, client *client) (sentryproto.Sentry_PeerEventsClient, error) { return client.PeerEvents(gctx, in, opts...) }), nil @@ -621,7 +623,7 @@ func (m *sentryMultiplexer) NodeInfo(ctx context.Context, in *emptypb.Empty, opt func (m *sentryMultiplexer) NodeInfos(ctx context.Context, opts ...grpc.CallOption) ([]*typesproto.NodeInfoReply, error) { var allInfos []*typesproto.NodeInfoReply - err := fanOut(ctx, m.clients, -1, func(gctx context.Context, client *client) (*typesproto.NodeInfoReply, error) { + 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)