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

beacon/goclient/aggregator.go

@@ -78,10 +78,21 @@
 
 // SubmitSignedAggregateSelectionProof broadcasts a signed aggregator msg
 func (gc *GoClient) SubmitSignedAggregateSelectionProof(msg *phase0.SignedAggregateAndProof) error {
+	clientAddress := gc.multiClient.Address()
+	logger := gc.log.With(
+		zap.String("api", "SubmitAggregateAttestations"),
+		zap.String("client_addr", clientAddress))
+
 	start := time.Now()
 	err := gc.multiClient.SubmitAggregateAttestations(gc.ctx, []*phase0.SignedAggregateAndProof{msg})
 	recordRequestDuration(gc.ctx, "SubmitAggregateAttestations", gc.multiClient.Address(), http.MethodPost, time.Since(start), err)
-	return err
+	if err != nil {
+		logger.Error(clResponseErrMsg, zap.Error(err))
+		return err
+	}
+
+	logger.Debug("consensus client submitted signed aggregate attestations")
+	return nil
 }
 
 // IsAggregator returns true if the signature is from the input validator. The committee

beacon/goclient/attest.go

@@ -122,16 +122,19 @@
 
 // SubmitAttestations implements Beacon interface
 func (gc *GoClient) SubmitAttestations(attestations []*phase0.Attestation) error {
+	clientAddress := gc.multiClient.Address()
+	logger := gc.log.With(
+		zap.String("api", "SubmitAttestations"),
+		zap.String("client_addr", clientAddress))
+
 	start := time.Now()
 	err := gc.multiClient.SubmitAttestations(gc.ctx, attestations)
-	recordRequestDuration(gc.ctx, "SubmitAttestations", gc.multiClient.Address(), http.MethodPost, time.Since(start), err)
+	recordRequestDuration(gc.ctx, "SubmitAttestations", clientAddress, http.MethodPost, time.Since(start), err)
 	if err != nil {
-		gc.log.Error(clResponseErrMsg,
-			zap.String("api", "SubmitAttestations"),
-			zap.Error(err),
-		)
+		logger.Error(clResponseErrMsg, zap.Error(err))
 		return err
 	}
 
-	return err
+	logger.Debug("consensus client submitted attestations")
+	return nil
 }

beacon/goclient/committee_subscribe.go

@@ -11,16 +11,21 @@
 
 // SubmitBeaconCommitteeSubscriptions is implementation for subscribing committee to subnet (p2p topic)
 func (gc *GoClient) SubmitBeaconCommitteeSubscriptions(ctx context.Context, subscription []*eth2apiv1.BeaconCommitteeSubscription) error {
+	clientAddress := gc.multiClient.Address()
+	logger := gc.log.With(
+		zap.String("api", "SubmitBeaconCommitteeSubscriptions"),
+		zap.String("client_addr", clientAddress))
+
 	start := time.Now()
 	err := gc.multiClient.SubmitBeaconCommitteeSubscriptions(ctx, subscription)
-	recordRequestDuration(gc.ctx, "SubmitBeaconCommitteeSubscriptions", gc.multiClient.Address(), http.MethodPost, time.Since(start), err)
+	recordRequestDuration(gc.ctx, "SubmitBeaconCommitteeSubscriptions", clientAddress, http.MethodPost, time.Since(start), err)
 	if err != nil {
-		gc.log.Error(clResponseErrMsg,
-			zap.String("api", "SubmitBeaconCommitteeSubscriptions"),
-			zap.Error(err),
-		)
+		logger.Error(clResponseErrMsg, zap.Error(err))
+		return err
 	}
-	return err
+
+	logger.Debug("consensus client submitted beacon committee subscriptions")
+	return nil
 }
 
 // SubmitSyncCommitteeSubscriptions is implementation for subscribing sync committee to subnet (p2p topic)

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.Named("consensus_client"),
+		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,131 @@
 	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()),
+		eth2clienthttp.WithSyncDistanceTolerance(gc.syncDistanceTolerance),
 	)
 	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.