diff --git a/beacon-chain/execution/BUILD.bazel b/beacon-chain/execution/BUILD.bazel index 2840c80b7d70..0b9bebe8dda9 100644 --- a/beacon-chain/execution/BUILD.bazel +++ b/beacon-chain/execution/BUILD.bazel @@ -101,8 +101,10 @@ go_test( "jsonrpc_error_test.go", "log_processing_test.go", "mock_test.go", + "options_test.go", "payload_body_test.go", "prometheus_test.go", + "rpc_connection_test.go", "service_test.go", ], data = glob(["testdata/**"]), diff --git a/beacon-chain/execution/options.go b/beacon-chain/execution/options.go index caaf72b5c5f7..413a96bd1532 100644 --- a/beacon-chain/execution/options.go +++ b/beacon-chain/execution/options.go @@ -45,6 +45,16 @@ func WithHttpEndpointAndJWTSecret(endpointString string, secret []byte) Option { } } +// WithRPCClientDialer supplies a dialer used to create the execution node RPC +// client (e.g. one from go-ethereum's rpc.DialInProc), taking precedence over +// any configured endpoint, JWT secret, and headers. +func WithRPCClientDialer(dialer RPCClientDialer) Option { + return func(s *Service) error { + s.cfg.rpcClientDialer = dialer + return nil + } +} + // WithHeaders adds headers to the execution node JSON-RPC requests. func WithHeaders(headers []string) Option { return func(s *Service) error { diff --git a/beacon-chain/execution/options_test.go b/beacon-chain/execution/options_test.go new file mode 100644 index 000000000000..8abd088caed2 --- /dev/null +++ b/beacon-chain/execution/options_test.go @@ -0,0 +1,26 @@ +package execution + +import ( + "context" + "testing" + + dbutil "github.com/OffchainLabs/prysm/v7/beacon-chain/db/testing" + "github.com/OffchainLabs/prysm/v7/testing/require" + "github.com/ethereum/go-ethereum/rpc" + "github.com/pkg/errors" +) + +func TestWithRPCClientDialer(t *testing.T) { + wantErr := errors.New("dialer invoked") + dialer := func(context.Context) (*rpc.Client, error) { + return nil, wantErr + } + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + require.NotNil(t, s.cfg.rpcClientDialer) + _, err = s.cfg.rpcClientDialer(t.Context()) + require.ErrorIs(t, err, wantErr) +} diff --git a/beacon-chain/execution/rpc_connection.go b/beacon-chain/execution/rpc_connection.go index cc3ef366a54b..adf58c16e3cc 100644 --- a/beacon-chain/execution/rpc_connection.go +++ b/beacon-chain/execution/rpc_connection.go @@ -18,21 +18,17 @@ import ( ) func (s *Service) setupExecutionClientConnections(ctx context.Context, currEndpoint network.Endpoint) error { - client, err := s.newRPCClientWithAuth(ctx, currEndpoint) + client, err := s.dialExecutionNode(ctx, currEndpoint) if err != nil { return errors.Wrap(err, "could not dial execution node") } - // Attach the clients to the service struct. fetcher := ethclient.NewClient(client) - s.rpcClient = client - s.httpLogger = fetcher depositContractCaller, err := contracts.NewDepositContractCaller(s.cfg.depositContractAddr, fetcher) if err != nil { client.Close() return errors.Wrap(err, "could not initialize deposit contract caller") } - s.depositContractCaller = depositContractCaller // Ensure we have the correct chain and deposit IDs. if err := ensureCorrectExecutionChain(ctx, fetcher); err != nil { @@ -45,6 +41,12 @@ func (s *Service) setupExecutionClientConnections(ctx context.Context, currEndpo } return errors.Wrap(err, errStr) } + + // Attach the clients to the service struct only after the connection is + // validated, so a failed attempt does not replace a working client. + s.rpcClient = client + s.httpLogger = fetcher + s.depositContractCaller = depositContractCaller s.updateConnectedETH1(true) s.runError = nil return nil @@ -64,10 +66,14 @@ func (s *Service) pollConnectionStatus(ctx context.Context) { } ticker := time.NewTicker(backOffPeriod) defer ticker.Stop() + dialTarget := logs.MaskCredentialsLogging(s.cfg.currHttpEndpoint.Url) + if s.cfg.rpcClientDialer != nil { + dialTarget = "injected RPC client dialer" + } for { select { case <-ticker.C: - log.Debugf("Trying to dial endpoint: %s", logs.MaskCredentialsLogging(s.cfg.currHttpEndpoint.Url)) + log.Debugf("Trying to dial endpoint: %s", dialTarget) currClient := s.rpcClient if err := s.setupExecutionClientConnections(ctx, s.cfg.currHttpEndpoint); err != nil { errorLogger(err, "Could not connect to execution client endpoint") @@ -77,7 +83,7 @@ func (s *Service) pollConnectionStatus(ctx context.Context) { if currClient != nil { currClient.Close() } - log.WithField("endpoint", logs.MaskCredentialsLogging(s.cfg.currHttpEndpoint.Url)).Info("Connected to new endpoint") + log.WithField("endpoint", dialTarget).Info("Connected to new endpoint") c, err := s.ExchangeCapabilities(ctx) if err != nil { @@ -119,6 +125,21 @@ func (s *Service) retryExecutionClientConnection(ctx context.Context, err error) s.runError = nil } +// Initializes the execution node RPC client, using the injected dialer when one is configured. +func (s *Service) dialExecutionNode(ctx context.Context, currEndpoint network.Endpoint) (*gethRPC.Client, error) { + if s.cfg.rpcClientDialer == nil { + return s.newRPCClientWithAuth(ctx, currEndpoint) + } + client, err := s.cfg.rpcClientDialer(ctx) + if err != nil { + return nil, err + } + if client == nil { + return nil, errors.New("rpc client dialer returned a nil client") + } + return client, nil +} + // Initializes an RPC connection with authentication headers. func (s *Service) newRPCClientWithAuth(ctx context.Context, endpoint network.Endpoint) (*gethRPC.Client, error) { headers := http.Header{} diff --git a/beacon-chain/execution/rpc_connection_test.go b/beacon-chain/execution/rpc_connection_test.go new file mode 100644 index 000000000000..fe44d5bcec41 --- /dev/null +++ b/beacon-chain/execution/rpc_connection_test.go @@ -0,0 +1,187 @@ +package execution + +import ( + "context" + "math/big" + "strconv" + "testing" + "time" + + dbutil "github.com/OffchainLabs/prysm/v7/beacon-chain/db/testing" + "github.com/OffchainLabs/prysm/v7/config/params" + "github.com/OffchainLabs/prysm/v7/testing/assert" + "github.com/OffchainLabs/prysm/v7/testing/require" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/rpc" + "github.com/pkg/errors" +) + +// inProcTestRPC serves eth_chainId and net_version for in-process dialer tests. +type inProcTestRPC struct { + chainID uint64 +} + +func (r *inProcTestRPC) ChainId(_ context.Context) *hexutil.Big { + return (*hexutil.Big)(new(big.Int).SetUint64(r.chainID)) +} + +func (r *inProcTestRPC) Version(_ context.Context) string { + return strconv.FormatUint(r.chainID, 10) +} + +// newInProcServer returns an in-process RPC server reporting the given chain ID. +func newInProcServer(t *testing.T, chainID uint64) *rpc.Server { + srv := rpc.NewServer() + api := &inProcTestRPC{chainID: chainID} + require.NoError(t, srv.RegisterName("eth", api)) + require.NoError(t, srv.RegisterName("net", api)) + t.Cleanup(srv.Stop) + return srv +} + +// inProcDialer returns a dialer backed by an in-process RPC server, plus an invocation counter. +func inProcDialer(t *testing.T) (RPCClientDialer, *int) { + srv := newInProcServer(t, params.BeaconConfig().DepositChainID) + calls := new(int) + dialer := func(_ context.Context) (*rpc.Client, error) { + *calls++ + return rpc.DialInProc(srv), nil + } + return dialer, calls +} + +func overrideBackOffPeriod(t *testing.T, d time.Duration) { + orig := backOffPeriod + backOffPeriod = d + t.Cleanup(func() { backOffPeriod = orig }) +} + +func TestSetupExecutionClientConnections_InjectedDialer(t *testing.T) { + dialer, calls := inProcDialer(t) + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + + // No endpoint is configured; the dialer alone must be enough to connect. + require.NoError(t, s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint)) + assert.Equal(t, 1, *calls) + assert.Equal(t, true, s.ExecutionClientConnected()) + assert.NotNil(t, s.depositContractCaller) + require.NoError(t, s.Stop()) +} + +func TestSetupExecutionClientConnections_DialerPrecedesEndpoint(t *testing.T) { + dialer, calls := inProcDialer(t) + // The configured endpoint accepts no connections; a successful setup proves + // the dialer took precedence over it. + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithHttpEndpoint("http://127.0.0.1:1"), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + + require.NoError(t, s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint)) + assert.Equal(t, 1, *calls) + assert.Equal(t, true, s.ExecutionClientConnected()) + require.NoError(t, s.Stop()) +} + +func TestSetupExecutionClientConnections_DialerReturnsNilClient(t *testing.T) { + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(func(context.Context) (*rpc.Client, error) { return nil, nil }), + ) + require.NoError(t, err) + + err = s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint) + require.ErrorContains(t, "nil client", err) + assert.Equal(t, false, s.ExecutionClientConnected()) +} + +func TestSetupExecutionClientConnections_FailedValidationKeepsPreviousClient(t *testing.T) { + goodServer := newInProcServer(t, params.BeaconConfig().DepositChainID) + badServer := newInProcServer(t, params.BeaconConfig().DepositChainID+1) + calls := 0 + dialer := func(_ context.Context) (*rpc.Client, error) { + calls++ + if calls == 2 { + return rpc.DialInProc(badServer), nil + } + return rpc.DialInProc(goodServer), nil + } + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + require.NoError(t, s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint)) + firstClient := s.rpcClient + + // The second dial reaches a node on the wrong chain: setup must fail without + // replacing or closing the previously attached client. + err = s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint) + require.ErrorContains(t, "wanted chain ID", err) + assert.Equal(t, true, firstClient == s.rpcClient, "previous client was replaced") + var res string + require.NoError(t, s.rpcClient.CallContext(t.Context(), &res, "net_version")) + + // A subsequent successful dial swaps the client as usual. + require.NoError(t, s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint)) + assert.Equal(t, 3, calls) + assert.Equal(t, true, firstClient != s.rpcClient, "client was not swapped after successful dial") + require.NoError(t, s.Stop()) +} + +func TestRetryExecutionClientConnection_ReinvokesDialer(t *testing.T) { + overrideBackOffPeriod(t, time.Millisecond) + dialer, calls := inProcDialer(t) + s, err := NewService(t.Context(), + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + require.NoError(t, s.setupExecutionClientConnections(t.Context(), s.cfg.currHttpEndpoint)) + prevClient := s.rpcClient + + s.retryExecutionClientConnection(t.Context(), errors.New("connection lost")) + + assert.Equal(t, 2, *calls) + assert.Equal(t, true, s.ExecutionClientConnected()) + require.NoError(t, s.runError) + // The reconnected client is usable and the previous client has been closed. + var res string + require.NoError(t, s.rpcClient.CallContext(t.Context(), &res, "net_version")) + err = prevClient.CallContext(t.Context(), &res, "net_version") + require.NotNil(t, err, "expected previous client to be closed") + require.NoError(t, s.Stop()) +} + +func TestPollConnectionStatus_InjectedDialerReconnects(t *testing.T) { + overrideBackOffPeriod(t, 5*time.Millisecond) + srv := newInProcServer(t, params.BeaconConfig().DepositChainID) + calls := 0 + dialer := func(_ context.Context) (*rpc.Client, error) { + calls++ + if calls < 3 { + return nil, errors.New("execution node not ready") + } + return rpc.DialInProc(srv), nil + } + // Bound the test so a broken poll loop fails fast instead of hanging. + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + s, err := NewService(ctx, + WithDatabase(dbutil.SetupDB(t)), + WithRPCClientDialer(dialer), + ) + require.NoError(t, err) + + // The poll loop keeps re-invoking the dialer until it succeeds. + s.pollConnectionStatus(ctx) + assert.Equal(t, 3, calls) + assert.Equal(t, true, s.ExecutionClientConnected()) + require.NoError(t, s.Stop()) +} diff --git a/beacon-chain/execution/service.go b/beacon-chain/execution/service.go index 0444aef312ea..f45459783d5e 100644 --- a/beacon-chain/execution/service.go +++ b/beacon-chain/execution/service.go @@ -118,6 +118,12 @@ func (RPCClientEmpty) CallContext(context.Context, any, string, ...any) error { return errors.New("rpc client is not initialized") } +// RPCClientDialer creates the RPC client used to communicate with the execution +// node. It is re-invoked on every reconnection attempt and must return a new, +// ready-to-use client on each call. Returned clients are owned and eventually +// closed by the service. +type RPCClientDialer func(ctx context.Context) (*gethRPC.Client, error) + // config defines a config struct for dependencies into the service. type config struct { depositContractAddr common.Address @@ -128,6 +134,7 @@ type config struct { eth1HeaderReqLimit uint64 beaconNodeStatsUpdater BeaconNodeStatsUpdater currHttpEndpoint network.Endpoint + rpcClientDialer RPCClientDialer headers []string finalizedStateAtStartup state.BeaconState jwtId string @@ -227,7 +234,7 @@ func (s *Service) Start() { } // If the chain has not started already and we don't have access to eth1 nodes, we will not be // able to generate the genesis state. - if !s.chainStartData.Chainstarted && s.cfg.currHttpEndpoint.Url == "" { + if !s.chainStartData.Chainstarted && s.cfg.currHttpEndpoint.Url == "" && s.cfg.rpcClientDialer == nil { // check for genesis state before shutting down the node, // if a genesis state exists, we can continue on. genState, err := s.cfg.beaconDB.GenesisState(s.ctx) diff --git a/changelog/satushh_execution-client-injection.md b/changelog/satushh_execution-client-injection.md new file mode 100644 index 000000000000..d941bbf0f7c0 --- /dev/null +++ b/changelog/satushh_execution-client-injection.md @@ -0,0 +1,7 @@ +### Added + +- Add `execution.WithRPCClientDialer` option allowing an embedding process to supply the execution node RPC client (e.g. one backed by `rpc.DialInProc`) instead of dialing the configured HTTP endpoint. + +### Fixed + +- Attach the execution service's RPC client only after chain ID validation succeeds, so a failed reconnection attempt no longer replaces a working client with a closed one or leaks the previous client.