Add WithRPCClientDialer option to inject the execution node RPC client

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/**"]),

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 {

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)
+}

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{}

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())
+}

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)

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.