Skip to content

node/gointerfaces/grpcutil: unify grpc server bring-up across daemons #22171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yperbasis merged 2 commits into from
Jul 3, 2026
Merged

node/gointerfaces/grpcutil: unify grpc server bring-up across daemons #22171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e17ddef
node/gointerfaces/grpcutil, cmd/downloader, txnprovider/txpool, node/…
yperbasis Jul 2, 2026
e2a850a
txnprovider/txpool, cmd/downloader, node/gointerfaces/grpcutil: take …
yperbasis Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 6 additions & 54 deletions cmd/downloader/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,29 +22,22 @@ import (
"errors"
"fmt"
"io/fs"
"net"
"net/url"
"os"
"path/filepath"
"runtime"
"slices"
"strings"
"time"

g "github.com/anacrolix/generics"
"github.com/anacrolix/missinggo/v2/panicif"
"github.com/anacrolix/torrent/metainfo"
"github.com/go-viper/mapstructure/v2"
"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
"github.com/pelletier/go-toml/v2"
"github.com/spf13/cobra"
"github.com/urfave/cli/v2"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/reflection"

"github.com/erigontech/erigon/cmd/utils/app"
"github.com/erigontech/erigon/db/downloader/webseeds"
Expand All @@ -67,6 +60,7 @@ import (
chainspec "github.com/erigontech/erigon/execution/chain/spec"
"github.com/erigontech/erigon/node/debug"
"github.com/erigontech/erigon/node/gointerfaces/downloaderproto"
"github.com/erigontech/erigon/node/gointerfaces/grpcutil"
"github.com/erigontech/erigon/node/logging"
"github.com/erigontech/erigon/node/paths"
"github.com/erigontech/erigon/p2p/nat"
Expand Down Expand Up @@ -761,57 +755,15 @@ func doDiffTorrentHashes(ctx context.Context, local map[string]string) error {
return nil
}

func StartGrpc(snServer *downloader.GrpcServer, addr string, creds *credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) {
lis, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr)
}

var (
streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 1)
unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 1)
)
streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor())
unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor())

//if metrics.Enabled {
// streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor)
// unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor)
//}

opts := []grpc.ServerOption{
// https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779
grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
MinTime: 10 * time.Second,
PermitWithoutStream: true,
}),
grpc.ChainStreamInterceptor(streamInterceptors...),
grpc.ChainUnaryInterceptor(unaryInterceptors...),
}
if creds == nil {
// no specific opts
} else {
opts = append(opts, grpc.Creds(*creds))
}
grpcServer := grpc.NewServer(opts...)
reflection.Register(grpcServer) // Register reflection service on gRPC server.
func StartGrpc(snServer *downloader.GrpcServer, addr string, creds credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) {
grpcServer := grpcutil.NewServerWithOpts(creds)
if snServer != nil {
downloaderproto.RegisterDownloaderServer(grpcServer, snServer)
}

//if metrics.Enabled {
// grpc_prometheus.Register(grpcServer)
//}

healthServer := health.NewServer()
grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)

go func() {
defer healthServer.Shutdown()
if err := grpcServer.Serve(lis); err != nil {
logger.Error("gRPC server stop", "err", err)
}
}()
if err := grpcutil.StartServer(grpcServer, addr, true, logger, "gRPC server stop"); err != nil {
return nil, err
}
logger.Info("Started gRPC server", "on", addr)
return grpcServer, nil
}
Expand Down
62 changes: 40 additions & 22 deletions node/gointerfaces/grpcutil/utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"errors"
"fmt"
"io"
"net"
"os"
"strings"
"time"
Expand All @@ -33,9 +34,13 @@ import (
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/reflection"
"google.golang.org/grpc/status"

"github.com/erigontech/erigon/common/log/v3"
)

func TLS(tlsCACert, tlsCertFile, tlsKeyFile string) (credentials.TransportCredentials, error) {
Expand Down Expand Up @@ -68,40 +73,53 @@ func TLS(tlsCACert, tlsCertFile, tlsKeyFile string) (credentials.TransportCreden
}

func NewServer(rateLimit uint32, creds credentials.TransportCredentials) *grpc.Server {
var (
streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 1)
unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 1)
)
streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor())
unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor())

//if metrics.Enabled {
// streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor)
// unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor)
//}
return NewServerWithOpts(creds, grpc.MaxConcurrentStreams(rateLimit)) // to force clients to reduce concurrency level
}
Comment thread
yperbasis marked this conversation as resolved.

//cpus := uint32(runtime.GOMAXPROCS(-1))
func NewServerWithOpts(creds credentials.TransportCredentials, extraOpts ...grpc.ServerOption) *grpc.Server {
opts := []grpc.ServerOption{
//grpc.NumStreamWorkers(cpus), // reduce amount of goroutines
grpc.MaxConcurrentStreams(rateLimit), // to force clients reduce concurrency level
// Don't drop the connection, settings accordign to this comment on GitHub
// Don't drop the connection, settings according to this comment on GitHub
// https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779
grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
MinTime: 10 * time.Second,
PermitWithoutStream: true,
}),
grpc.ChainStreamInterceptor(streamInterceptors...),
grpc.ChainUnaryInterceptor(unaryInterceptors...),
grpc.Creds(creds),
grpc.ChainStreamInterceptor(recovery.StreamServerInterceptor()),
grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor()),
}
if creds != nil {
opts = append(opts, grpc.Creds(creds))
}
opts = append(opts, extraOpts...)

grpcServer := grpc.NewServer(opts...)
reflection.Register(grpcServer)
return grpcServer
}

//if metrics.Enabled {
// grpc_prometheus.Register(grpcServer)
//}
func StartServer(srv *grpc.Server, addr string, healthCheck bool, logger log.Logger, serveErrMsg string) error {
lis, err := net.Listen("tcp", addr)
if err != nil {
return fmt.Errorf("could not create listener: %w, addr=%s", err, addr)
}
StartServerOnListener(srv, lis, healthCheck, logger, serveErrMsg)
return nil
}

return grpcServer
func StartServerOnListener(srv *grpc.Server, lis net.Listener, healthCheck bool, logger log.Logger, serveErrMsg string) {
var healthServer *health.Server
if healthCheck {
healthServer = health.NewServer()
grpc_health_v1.RegisterHealthServer(srv, healthServer)
}
go func() {
if healthServer != nil {
defer healthServer.Shutdown()
}
if err := srv.Serve(lis); err != nil {
logger.Error(serveErrMsg, "err", err)
}
}()
Comment thread
yperbasis marked this conversation as resolved.
}

func Connect(creds credentials.TransportCredentials, dialAddress string) (*grpc.ClientConn, error) {
Expand Down
26 changes: 3 additions & 23 deletions node/privateapi/all.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,8 @@
package privateapi

import (
"fmt"
"net"

"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"

"github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/db/kv/remotedbserver"
Expand All @@ -38,10 +33,6 @@ func StartGrpc(kv *remotedbserver.KvServer, ethBackendSrv *EthBackendServer, txP
miningServer txpoolproto.MiningServer, bridgeServer *bridge.BackendServer, heimdallServer *heimdall.BackendServer,
addr string, rateLimit uint32, creds credentials.TransportCredentials, healthCheck bool, logger log.Logger) (*grpc.Server, error) {
logger.Info("Starting private RPC server", "on", addr)
lis, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr)
}

grpcServer := grpcutil.NewServer(rateLimit, creds)
remoteproto.RegisterETHBACKENDServer(grpcServer, ethBackendSrv)
Expand All @@ -57,21 +48,10 @@ func StartGrpc(kv *remotedbserver.KvServer, ethBackendSrv *EthBackendServer, txP
if heimdallServer != nil {
remoteproto.RegisterHeimdallBackendServer(grpcServer, heimdallServer)
}

remoteproto.RegisterKVServer(grpcServer, kv)
var healthServer *health.Server
if healthCheck {
healthServer = health.NewServer()
grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)
}
go func() {
if healthCheck {
defer healthServer.Shutdown()
}
if err := grpcServer.Serve(lis); err != nil {
logger.Error("private RPC server fail", "err", err)
}
}()

if err := grpcutil.StartServer(grpcServer, addr, healthCheck, logger, "private RPC server fail"); err != nil {
return nil, err
}
return grpcServer, nil
}
17 changes: 1 addition & 16 deletions p2p/sentry/sentry_grpc_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,8 +40,6 @@ import (
"golang.org/x/sync/errgroup"
"golang.org/x/time/rate"
"google.golang.org/grpc"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/protobuf/types/known/emptypb"

"github.com/erigontech/erigon/common"
Expand Down Expand Up @@ -887,20 +885,7 @@ func grpcSentryServer(ctx context.Context, sentryAddr string, ss *GrpcServer, he
}
grpcServer := grpcutil.NewServer(100, nil)
sentryproto.RegisterSentryServer(grpcServer, ss)
var healthServer *health.Server
if healthCheck {
healthServer = health.NewServer()
grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)
}

go func() {
if healthCheck {
defer healthServer.Shutdown()
}
if err1 := grpcServer.Serve(lis); err1 != nil {
ss.logger.Error("Sentry gRPC server fail", "err", err1)
}
}()
grpcutil.StartServerOnListener(grpcServer, lis, healthCheck, ss.logger, "Sentry gRPC server fail")
return grpcServer, nil
}

Expand Down
64 changes: 6 additions & 58 deletions txnprovider/txpool/txpool_grpc_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,19 +19,11 @@ package txpool
import (
"context"
"errors"
"fmt"
"math"
"net"
"time"

"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
"github.com/holiman/uint256"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/reflection"
"google.golang.org/protobuf/types/known/emptypb"

"github.com/erigontech/erigon/common"
Expand Down Expand Up @@ -325,65 +317,21 @@ func (s *GrpcServer) Nonce(ctx context.Context, in *txpoolproto.NonceRequest) (*
// NewSlotsStreams - it's safe to use this class as non-pointer
type NewSlotsStreams = grpcutil.StreamBroadcaster[txpoolproto.OnAddReply]

func StartGrpc(txPoolServer txpoolproto.TxpoolServer, miningServer txpoolproto.MiningServer, addr string, creds *credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) {
lis, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr)
}

var (
streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 2)
unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 2)
)
streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor())
unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor())

//if metrics.Enabled {
// streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor)
// unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor)
//}

//cpus := uint32(runtime.GOMAXPROCS(-1))
opts := []grpc.ServerOption{
//grpc.NumStreamWorkers(cpus), // reduce amount of goroutines
func StartGrpc(txPoolServer txpoolproto.TxpoolServer, miningServer txpoolproto.MiningServer, addr string, creds credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) {
grpcServer := grpcutil.NewServerWithOpts(creds,
grpc.ReadBufferSize(0), // reduce buffers to save mem
grpc.WriteBufferSize(0), // reduce buffers to save mem
// Don't drop the connection, settings accordign to this comment on GitHub
// https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779
grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
MinTime: 10 * time.Second,
PermitWithoutStream: true,
}),
grpc.ChainStreamInterceptor(streamInterceptors...),
grpc.ChainUnaryInterceptor(unaryInterceptors...),
}
if creds == nil {
// no specific opts
} else {
opts = append(opts, grpc.Creds(*creds))
}
grpcServer := grpc.NewServer(opts...)
reflection.Register(grpcServer) // Register reflection service on gRPC server.
)
if txPoolServer != nil {
txpoolproto.RegisterTxpoolServer(grpcServer, txPoolServer)
}
if miningServer != nil {
txpoolproto.RegisterMiningServer(grpcServer, miningServer)
}

//if metrics.Enabled {
// grpc_prometheus.Register(grpcServer)
//}

healthServer := health.NewServer()
grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)

go func() {
defer healthServer.Shutdown()
if err := grpcServer.Serve(lis); err != nil {
logger.Error("private RPC server fail", "err", err)
}
}()
if err := grpcutil.StartServer(grpcServer, addr, true, logger, "txpool gRPC server fail"); err != nil {
return nil, err
}
logger.Info("Started gRPC server", "on", addr)
return grpcServer, nil
}
Loading