hotfix(el/cl): allow multi clients to start if at least one node is up

beacon/goclient/goclient.go

@@ -6,6 +6,7 @@
 	"math"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	eth2client "github.com/attestantio/go-eth2-client"
@@ -114,6 +115,8 @@
 	clients     []Client
 	multiClient MultiClient
 
+	genesisVersion atomic.Pointer[phase0.Version]
+
 	syncDistanceTolerance phase0.Slot
 	nodeSyncingFn         func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*apiv1.SyncState], error)
 
@@ -154,82 +157,41 @@
 		longTimeout = DefaultLongTimeout
 	}
 
+	client := &GoClient{
+		log:                   logger,
+		ctx:                   opt.Context,
+		network:               opt.Network,
+		syncDistanceTolerance: phase0.Slot(opt.SyncDistanceTolerance),
+		operatorDataStore:     operatorDataStore,
+		registrationCache:     map[phase0.BLSPubKey]*api.VersionedSignedValidatorRegistration{},
+		attestationDataCache: ttlcache.New(
+			// we only fetch attestation data during the slot of the relevant duty (and never later),
+			// hence caching it for 2 slots is sufficient
+			ttlcache.WithTTL[phase0.Slot, *phase0.AttestationData](2 * opt.Network.SlotDurationSec()),
+		),
+		commonTimeout: commonTimeout,
+		longTimeout:   longTimeout,
+	}
+
 	beaconAddrList := strings.Split(opt.BeaconNodeAddr, ";") // TODO: Decide what symbol to use as a separator. Bootnodes are currently separated by ";". Deployment bot currently uses ",".
 	if len(beaconAddrList) == 0 {
 		return nil, fmt.Errorf("no beacon node address provided")
 	}
 
-	var consensusClients []Client
-	var consensusClientsAsServices []eth2client.Service
 	for _, beaconAddr := range beaconAddrList {
-		httpClient, err := setupHTTPClient(opt.Context, logger, beaconAddr, commonTimeout)
-		if err != nil {
+		if err := client.addSingleClient(opt.Context, beaconAddr); err != nil {
 			return nil, err
 		}
-
-		nodeVersionResp, err := httpClient.NodeVersion(opt.Context, &api.NodeVersionOpts{})
-		if err != nil {
-			logger.Error(clResponseErrMsg,
-				zap.String("api", "NodeVersion"),
-				zap.Error(err),
-			)
-			return nil, fmt.Errorf("failed to get node version: %w", err)
-		}
-		if nodeVersionResp == nil {
-			logger.Error(clNilResponseErrMsg,
-				zap.String("api", "NodeVersion"),
-			)
-			return nil, fmt.Errorf("node version response is nil")
-		}
-
-		logger.Info("consensus client connected",
-			fields.Name(httpClient.Name()),
-			fields.Address(httpClient.Address()),
-			zap.String("client", string(ParseNodeClient(nodeVersionResp.Data))),
-			zap.String("version", nodeVersionResp.Data),
-		)
-
-		consensusClients = append(consensusClients, httpClient)
-		consensusClientsAsServices = append(consensusClientsAsServices, httpClient)
 	}
 
-	err := assertSameGenesis(opt.Context, consensusClients...)
-	if err != nil {
-		return nil, fmt.Errorf("assert same spec: %w", err)
-	}
-
-	multiClient, err := eth2clientmulti.New(
-		opt.Context,
-		eth2clientmulti.WithClients(consensusClientsAsServices),
-		eth2clientmulti.WithLogLevel(zerolog.DebugLevel),
-		eth2clientmulti.WithTimeout(commonTimeout),
-	)
+	err := client.initMultiClient(opt.Context)
 	if err != nil {
 		logger.Error("Consensus multi client initialization failed",
 			zap.String("address", opt.BeaconNodeAddr),
 			zap.Error(err),
 		)
 
-		return nil, fmt.Errorf("create multi client: %w", err)
-	}
-	consensusClient := multiClient.(*eth2clientmulti.Service)
-
-	client := &GoClient{
-		log:                   logger,
-		ctx:                   opt.Context,
-		network:               opt.Network,
-		clients:               consensusClients,
-		multiClient:           consensusClient,
-		syncDistanceTolerance: phase0.Slot(opt.SyncDistanceTolerance),
-		operatorDataStore:     operatorDataStore,
-		registrationCache:     map[phase0.BLSPubKey]*api.VersionedSignedValidatorRegistration{},
-		attestationDataCache: ttlcache.New(
-			// we only fetch attestation data during the slot of the relevant duty (and never later),
-			// hence caching it for 2 slots is sufficient
-			ttlcache.WithTTL[phase0.Slot, *phase0.AttestationData](2 * opt.Network.SlotDurationSec()),
-		),
-		commonTimeout: commonTimeout,
-		longTimeout:   longTimeout,
+		return nil, err
 	}
 
 	client.nodeSyncingFn = client.nodeSyncing
@@ -241,68 +203,130 @@
 	return client, nil
 }
 
-func setupHTTPClient(ctx context.Context, logger *zap.Logger, addr string, commonTimeout time.Duration) (*eth2clienthttp.Service, error) {
+func (gc *GoClient) initMultiClient(ctx context.Context) error {
+	var services []eth2client.Service
+	for _, client := range gc.clients {
+		services = append(services, client)
+	}
+
+	multiClient, err := eth2clientmulti.New(
+		ctx,
+		eth2clientmulti.WithClients(services),
+		eth2clientmulti.WithLogLevel(zerolog.DebugLevel),
+		eth2clientmulti.WithTimeout(gc.commonTimeout),
+	)
+	if err != nil {
+		return fmt.Errorf("create multi client: %w", err)
+	}
+
+	gc.multiClient = multiClient.(*eth2clientmulti.Service)
+	return nil
+}
+
+func (gc *GoClient) addSingleClient(ctx context.Context, addr string) error {
 	httpClient, err := eth2clienthttp.New(
 		ctx,
 		// WithAddress supplies the address of the beacon node, in host:port format.
 		eth2clienthttp.WithAddress(addr),
 		// LogLevel supplies the level of logging to carry out.
 		eth2clienthttp.WithLogLevel(zerolog.DebugLevel),
-		eth2clienthttp.WithTimeout(commonTimeout),
+		eth2clienthttp.WithTimeout(gc.commonTimeout),
 		eth2clienthttp.WithReducedMemoryUsage(true),
 		eth2clienthttp.WithAllowDelayedStart(true),
+		eth2clienthttp.WithHooks(gc.singleClientHooks()),
 	)
 	if err != nil {
-		logger.Error("Consensus http client initialization failed",
+		gc.log.Error("Consensus http client initialization failed",
 			zap.String("address", addr),
 			zap.Error(err),
 		)
 
-		return nil, fmt.Errorf("create http client: %w", err)
+		return fmt.Errorf("create http client: %w", err)
 	}
 
-	return httpClient.(*eth2clienthttp.Service), nil
-}
-
-// assertSameGenesis should receive a non-empty list
-func assertSameGenesis(ctx context.Context, services ...Client) error {
-	firstGenesis, err := services[0].Genesis(ctx, &api.GenesisOpts{})
-	if err != nil {
-		return fmt.Errorf("get first genesis: %w", err)
-	}
-
-	for _, service := range services[1:] {
-		srvGenesis, err := service.Genesis(ctx, &api.GenesisOpts{})
-		if err != nil {
-			return fmt.Errorf("get service genesis: %w", err)
-		}
-
-		if err := sameGenesis(firstGenesis.Data, srvGenesis.Data); err != nil {
-			return fmt.Errorf("different genesis: %w", err)
-		}
-	}
+	gc.clients = append(gc.clients, httpClient.(*eth2clienthttp.Service))
 
 	return nil
 }
 
-func sameGenesis(a, b *apiv1.Genesis) error {
-	if a == nil || b == nil { // Input parameters should never be nil, so the check may fail if both are nil
-		return fmt.Errorf("genesis is nil")
-	}
+func (gc *GoClient) singleClientHooks() *eth2clienthttp.Hooks {
+	return &eth2clienthttp.Hooks{
+		OnActive: func(ctx context.Context, s *eth2clienthttp.Service) {
+			// If err is nil, nodeVersionResp is never nil.
+			nodeVersionResp, err := s.NodeVersion(ctx, &api.NodeVersionOpts{})
+			if err != nil {
+				gc.log.Error(clResponseErrMsg,
+					zap.String("address", s.Address()),
+					zap.String("api", "NodeVersion"),
+					zap.Error(err),
+				)
+				return
+			}
+
+			gc.log.Info("consensus client connected",
+				fields.Name(s.Name()),
+				fields.Address(s.Address()),
+				zap.String("client", string(ParseNodeClient(nodeVersionResp.Data))),
+				zap.String("version", nodeVersionResp.Data),
+			)
 
-	if !a.GenesisTime.Equal(b.GenesisTime) {
-		return fmt.Errorf("genesis time mismatch, got %v and %v", a.GenesisTime, b.GenesisTime)
+			genesis, err := s.Genesis(ctx, &api.GenesisOpts{})
+			if err != nil {
+				gc.log.Error(clResponseErrMsg,
+					zap.String("address", s.Address()),
+					zap.String("api", "Genesis"),
+					zap.Error(err),
+				)
+				return
+			}
+
+			if expected, err := gc.assertSameGenesisVersion(genesis.Data.GenesisForkVersion); err != nil {
+				gc.log.Fatal("client returned unexpected genesis fork version, make sure all clients use the same Ethereum network",
+					zap.String("address", s.Address()),
+					zap.Any("client_genesis", genesis.Data.GenesisForkVersion),
+					zap.Any("expected_genesis", expected),
+					zap.Error(err),
+				)
+				return // Tests may override Fatal's behavior
+			}
+		},
+		OnInactive: func(ctx context.Context, s *eth2clienthttp.Service) {
+			gc.log.Warn("consensus client disconnected",
+				fields.Name(s.Name()),
+				fields.Address(s.Address()),
+			)
+		},
+		OnSynced: func(ctx context.Context, s *eth2clienthttp.Service) {
+			gc.log.Info("consensus client synced",
+				fields.Name(s.Name()),
+				fields.Address(s.Address()),
+			)
+		},
+		OnDesynced: func(ctx context.Context, s *eth2clienthttp.Service) {
+			gc.log.Warn("consensus client desynced",
+				fields.Name(s.Name()),
+				fields.Address(s.Address()),
+			)
+		},
 	}
+}
 
-	if a.GenesisValidatorsRoot != b.GenesisValidatorsRoot {
-		return fmt.Errorf("genesis validators root mismatch, got %v and %v", a.GenesisValidatorsRoot, b.GenesisValidatorsRoot)
+// assertSameGenesis checks if genesis is same.
+// Clients may have different values returned by Spec call,
+// so we decided that it's best to assert that GenesisForkVersion is the same.
+// To add more assertions, we check the whole apiv1.Genesis (GenesisTime and GenesisValidatorsRoot)
+// as they should be same too.
+func (gc *GoClient) assertSameGenesisVersion(genesisVersion phase0.Version) (phase0.Version, error) {
+	if gc.genesisVersion.CompareAndSwap(nil, &genesisVersion) {
+		return genesisVersion, nil
 	}
 
-	if a.GenesisForkVersion != b.GenesisForkVersion {
-		return fmt.Errorf("genesis fork version mismatch, got %v and %v", a.GenesisForkVersion, b.GenesisForkVersion)
+	expected := *gc.genesisVersion.Load()
+	if expected != genesisVersion {
+		return expected, fmt.Errorf("genesis fork version mismatch, expected %v, got %v", expected, genesisVersion)
 	}
 
-	return nil
+	return expected, nil
 }
 
 func (gc *GoClient) nodeSyncing(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*apiv1.SyncState], error) {

beacon/goclient/goclient_test.go

@@ -88,7 +88,7 @@ func TestTimeouts(t *testing.T) {
 			return nil
 		})
 		_, err := mockClient(ctx, undialableServer.URL, commonTimeout, longTimeout)
-		require.ErrorContains(t, err, "context deadline exceeded")
+		require.ErrorContains(t, err, "client is not active")
 	}
 
 	// Too slow to respond to the Validators request.

eth/executionclient/config.go

@@ -8,7 +8,7 @@ import (
 
 // ExecutionOptions contains config configurations related to Ethereum execution client.
 type ExecutionOptions struct {
-	Addr                  string        `yaml:"ETH1Addr" env:"ETH_1_ADDR" env-required:"true" env-description:"Execution client WebSocket address. Supports multiple comma-separated addresses"`
+	Addr                  string        `yaml:"ETH1Addr" env:"ETH_1_ADDR" env-required:"true" env-description:"Execution client WebSocket address. Supports multiple semicolon separated addresses. ex: ws://localhost:8546;ws://localhost:8547"`
 	ConnectionTimeout     time.Duration `yaml:"ETH1ConnectionTimeout" env:"ETH_1_CONNECTION_TIMEOUT" env-default:"10s" env-description:"Execution client connection timeout"`
 	SyncDistanceTolerance uint64        `yaml:"ETH1SyncDistanceTolerance" env:"ETH_1_SYNC_DISTANCE_TOLERANCE" env-default:"5" env-description:"The number of out-of-sync blocks we can tolerate"`
 }

eth/executionclient/execution_client.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -49,7 +48,6 @@ var _ Provider = &ExecutionClient{}
 
 var (
 	ErrClosed        = fmt.Errorf("closed")
-	ErrUnhealthy     = fmt.Errorf("unhealthy")
 	ErrNotConnected  = fmt.Errorf("not connected")
 	ErrBadInput      = fmt.Errorf("bad input")
 	ErrNothingToSync = errors.New("nothing to sync")
@@ -81,8 +79,6 @@ type ExecutionClient struct {
 	client         *ethclient.Client
 	closed         chan struct{}
 	lastSyncedTime atomic.Int64
-	healthyChMu    sync.Mutex
-	healthyCh      chan struct{}
 }
 
 // New creates a new instance of ExecutionClient.
@@ -98,7 +94,6 @@ func New(ctx context.Context, nodeAddr string, contractAddr ethcommon.Address, o
 		healthInvalidationInterval:  DefaultHealthInvalidationInterval,
 		logBatchSize:                DefaultHistoricalLogsBatchSize, // TODO Make batch of logs adaptive depending on "websocket: read limit"
 		closed:                      make(chan struct{}),
-		healthyCh:                   make(chan struct{}),
 	}
 	for _, opt := range opts {
 		opt(client)
@@ -246,8 +241,6 @@ func (ec *ExecutionClient) StreamLogs(ctx context.Context, fromBlock uint64) <-c
 				return
 			case <-ec.closed:
 				return
-			case <-ec.healthyCh:
-				return
 			default:
 				lastBlock, err := ec.streamLogsToChan(ctx, logs, fromBlock)
 				if errors.Is(err, ErrClosed) || errors.Is(err, context.Canceled) {
@@ -294,22 +287,7 @@ func (ec *ExecutionClient) Healthy(ctx context.Context) error {
 		return nil
 	}
 
-	ec.healthyChMu.Lock()
-	defer ec.healthyChMu.Unlock()
-
-	if err := ec.healthy(ctx); err != nil {
-		close(ec.healthyCh)
-		return fmt.Errorf("unhealthy: %w", err)
-	}
-
-	// Reset the healthyCh channel if it was closed.
-	select {
-	case <-ec.healthyCh:
-	default:
-		ec.healthyCh = make(chan struct{})
-	}
-
-	return nil
+	return ec.healthy(ctx)
 }
 
 func (ec *ExecutionClient) healthy(ctx context.Context) error {
@@ -426,9 +404,6 @@ func (ec *ExecutionClient) streamLogsToChan(ctx context.Context, logs chan<- Blo
 		case <-ec.closed:
 			return fromBlock, ErrClosed
 
-		case <-ec.healthyCh:
-			return fromBlock, ErrUnhealthy
-
 		case err := <-sub.Err():
 			if err == nil {
 				return fromBlock, ErrClosed