fix(operator): manage subscription to events without panic (#1729)

Co-authored-by: Urix <[email protected]>
Co-authored-by: Mauro Toscano <[email protected]>
Co-authored-by: Marcos Nicolau <[email protected]>
Co-authored-by: Marcos Nicolau <[email protected]>

Author: 4 people

aggregator/cmd/main.go

@@ -60,9 +60,9 @@ func aggregatorMain(ctx *cli.Context) error {
 
 	// Listen for new task created in the ServiceManager contract in a separate goroutine, both V1 and V2 subscriptions:
 	go func() {
-		listenErr := aggregator.SubscribeToNewTasks()
-		if listenErr != nil {
-			aggregatorConfig.BaseConfig.Logger.Fatal("Error subscribing for new tasks", "err", listenErr)
+		listenErrPair := aggregator.SubscribeToNewTasks()
+		if listenErrPair != nil {
+			aggregatorConfig.BaseConfig.Logger.Fatal("Error subscribing for new tasks", "err", listenErrPair)
 		}
 	}()
 

aggregator/pkg/aggregator.go

@@ -46,7 +46,7 @@ type Aggregator struct {
 	avsReader             *chainio.AvsReader
 	avsSubscriber         *chainio.AvsSubscriber
 	avsWriter             *chainio.AvsWriter
-	taskSubscriber        chan error
+	taskSubscriber        chan chainio.ErrorPair
 	blsAggregationService blsagg.BlsAggregationService
 
 	// BLS Signature Service returns an Index

aggregator/pkg/subscriber.go

@@ -1,18 +1,20 @@
 package pkg
 
-func (agg *Aggregator) SubscribeToNewTasks() error {
-	err := agg.subscribeToNewTasks()
-	if err != nil {
-		return err
+import "github.com/yetanotherco/aligned_layer/core/chainio"
+
+func (agg *Aggregator) SubscribeToNewTasks() *chainio.ErrorPair {
+	errorPair := agg.subscribeToNewTasks()
+	if errorPair != nil {
+		return errorPair
 	}
 
 	for {
 		select {
 		case err := <-agg.taskSubscriber:
 			agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to subscribe to new tasks", "err", err)
-			err = agg.subscribeToNewTasks()
-			if err != nil {
-				return err
+			errorPair = agg.subscribeToNewTasks()
+			if errorPair != nil {
+				return errorPair
 			}
 		case newBatch := <-agg.NewBatchChan:
 			agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task")
@@ -21,14 +23,12 @@ func (agg *Aggregator) SubscribeToNewTasks() error {
 	}
 }
 
-func (agg *Aggregator) subscribeToNewTasks() error {
-	var err error
-
-	agg.taskSubscriber, err = agg.avsSubscriber.SubscribeToNewTasksV3(agg.NewBatchChan)
+func (agg *Aggregator) subscribeToNewTasks() *chainio.ErrorPair {
+	errorPair := agg.avsSubscriber.SubscribeToNewTasksV3(agg.NewBatchChan, agg.taskSubscriber)
 
-	if err != nil {
-		agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to create task subscriber", "err", err)
+	if errorPair != nil {
+		agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to create task subscriber", "err", errorPair)
 	}
 
-	return err
+	return errorPair
 }

config-files/config-operator-1-ethereum-package.yaml

@@ -3,10 +3,10 @@
 environment: 'development'
 aligned_layer_deployment_config_file_path: './contracts/script/output/devnet/alignedlayer_deployment_output.json'
 eigen_layer_deployment_config_file_path: './contracts/script/output/devnet/eigenlayer_deployment_output.json'
-eth_rpc_url: "http://localhost:8545"
-eth_rpc_url_fallback: "http://localhost:8552"
-eth_ws_url: "ws://localhost:8546"
-eth_ws_url_fallback: "ws://localhost:8553"
+eth_rpc_url: "http://localhost:8552"
+eth_rpc_url_fallback: "http://localhost:8559"
+eth_ws_url: "ws://localhost:8553"
+eth_ws_url_fallback: "ws://localhost:8560"
 eigen_metrics_ip_port_address: 'localhost:9090'
 
 ## ECDSA Configurations

core/chainio/avs_reader.go

@@ -72,7 +72,15 @@ func (r *AvsReader) IsOperatorRegistered(address ethcommon.Address) (bool, error
 }
 
 func (r *AvsReader) DisabledVerifiers() (*big.Int, error) {
-	return r.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.DisabledVerifiers(&bind.CallOpts{})
+	num, err := r.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.DisabledVerifiers(&bind.CallOpts{})
+	if err != nil {
+		// Retry with fallback client
+		num, err = r.AvsContractBindings.ServiceManagerFallback.ContractAlignedLayerServiceManagerCaller.DisabledVerifiers(&bind.CallOpts{})
+		if err != nil {
+			r.logger.Error("Failed to fetch DisabledVerifiers", "err", err)
+		}
+	}
+	return num, err
 }
 
 // Returns all the "NewBatchV3" logs that have not been responded starting from the given block number

core/chainio/avs_subscriber.go

@@ -8,12 +8,14 @@ import (
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	ethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/event"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
 	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/config"
 
+	"fmt"
 	sdklogging "github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/ethereum/go-ethereum/crypto"
 )
@@ -43,6 +45,11 @@ type AvsSubscriber struct {
 	logger                         sdklogging.Logger
 }
 
+type ErrorPair struct {
+	ErrorMainRPC     error
+	ErrorFallbackRPC error
+}
+
 func NewAvsSubscriberFromConfig(baseConfig *config.BaseConfig) (*AvsSubscriber, error) {
 	avsContractBindings, err := NewAvsServiceBindings(
 		baseConfig.AlignedLayerDeploymentConfig.AlignedLayerServiceManagerAddr,
@@ -61,26 +68,27 @@ func NewAvsSubscriberFromConfig(baseConfig *config.BaseConfig) (*AvsSubscriber,
 	}, nil
 }
 
-func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) (chan error, error) {
+func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, errorPairChannel chan ErrorPair) *ErrorPair {
 	// Create a new channel to receive new tasks
 	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
 
 	// Subscribe to new tasks
-	sub, err := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-	if err != nil {
-		s.logger.Error("Primary failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.NetworkNumRetries, "err", err)
-		return nil, err
+	sub, errMain := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
+	if errMain != nil {
+		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "errMain", fmt.Sprintf("%v", errMain))
 	}
 
-	subFallback, err := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-	if err != nil {
-		s.logger.Error("Fallback failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.NetworkNumRetries, "err", err)
-		return nil, err
+	subFallback, errFallback := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
+	if errFallback != nil {
+		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "errFallback", fmt.Sprintf("%v", errFallback))
 	}
-	s.logger.Info("Subscribed to new AlignedLayer V2 tasks")
 
-	// create a new channel to foward errors
-	errorChannel := make(chan error)
+	if errMain != nil && errFallback != nil {
+		s.logger.Error("Failed to subscribe to new AlignedLayer V2 tasks with both RPCs", "errMain", errMain, "errFallback", errFallback)
+		return &ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
+	}
+
+	s.logger.Info("Subscribed to new AlignedLayer V2 tasks")
 
 	pollLatestBatchTicker := time.NewTicker(PollLatestBatchInterval)
 
@@ -109,49 +117,56 @@ func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicema
 
 	// Handle errors and resubscribe
 	go func() {
-		for {
+		var errMain, errFallback error
+		var auxSub, auxSubFallback event.Subscription
+		for errMain == nil || errFallback == nil { //while one is active
 			select {
 			case err := <-sub.Err():
-				s.logger.Warn("Error in new task subscription", "err", err)
-				sub.Unsubscribe()
-				sub, err = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-				if err != nil {
-					errorChannel <- err
+				s.logger.Warn("Error in new task subscription of main connection", "err", err)
+
+				auxSub, errMain = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
+				if errMain == nil {
+					sub = auxSub // update the subscription only if it was successful
+					s.logger.Info("Main connection resubscribed to new task subscription")
 				}
 			case err := <-subFallback.Err():
-				s.logger.Warn("Error in fallback new task subscription", "err", err)
-				subFallback.Unsubscribe()
-				subFallback, err = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-				if err != nil {
-					errorChannel <- err
+				s.logger.Warn("Error in new task subscription of fallback connection", "err", err)
+
+				auxSubFallback, errFallback = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
+				if errFallback == nil {
+					subFallback = auxSubFallback // update the subscription only if it was successful
+					s.logger.Info("Resubscribed to fallback new task subscription")
 				}
 			}
 		}
+		errorPairChannel <- ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
 	}()
 
-	return errorChannel, nil
+	return nil
 }
 
-func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3) (chan error, error) {
+func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, errorPairChannel chan ErrorPair) *ErrorPair {
 	// Create a new channel to receive new tasks
 	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3)
 
+	s.logger.Info("Starting subscription to new AlignedLayer V3 tasks")
 	// Subscribe to new tasks
-	sub, err := SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-	if err != nil {
-		s.logger.Error("Primary failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries, "err", err)
-		return nil, err
+	sub, errMain := SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
+	if errMain != nil {
+		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "err", fmt.Sprintf("%v", errMain))
 	}
 
-	subFallback, err := SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-	if err != nil {
-		s.logger.Error("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries, "err", err)
-		return nil, err
+	subFallback, errFallback := SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
+	if errFallback != nil {
+		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "err", fmt.Sprintf("%v", errFallback))
 	}
-	s.logger.Info("Subscribed to new AlignedLayer V3 tasks")
 
-	// create a new channel to foward errors
-	errorChannel := make(chan error)
+	if errMain != nil && errFallback != nil {
+		s.logger.Error("Failed to subscribe to new AlignedLayer V3 tasks with both RPCs", "errMain", errMain, "errFallback", errFallback)
+		return &ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
+	}
+
+	s.logger.Info("Subscribed to new AlignedLayer V3 tasks")
 
 	pollLatestBatchTicker := time.NewTicker(PollLatestBatchInterval)
 
@@ -180,27 +195,33 @@ func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicema
 
 	// Handle errors and resubscribe
 	go func() {
-		for {
+		s.logger.Info("Starting error handling goroutine")
+		var errMain, errFallback error
+		var auxSub, auxSubFallback event.Subscription
+		for errMain == nil || errFallback == nil { //while one is active
 			select {
 			case err := <-sub.Err():
-				s.logger.Warn("Error in new task subscription", "err", err)
-				sub.Unsubscribe()
-				sub, err = SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-				if err != nil {
-					errorChannel <- err
+				s.logger.Warn("Error in new task subscription of main connection", "err", err)
+
+				auxSub, errMain = SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
+				if errMain == nil {
+					sub = auxSub // update the subscription only if it was successful
+					s.logger.Info("Resubscribed to fallback new task subscription")
 				}
 			case err := <-subFallback.Err():
-				s.logger.Warn("Error in fallback new task subscription", "err", err)
-				subFallback.Unsubscribe()
-				subFallback, err = SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-				if err != nil {
-					errorChannel <- err
+				s.logger.Warn("Error in new task subscription of fallback connection", "err", err)
+
+				auxSubFallback, errFallback = SubscribeToNewTasksV3Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
+				if errFallback == nil {
+					subFallback = auxSubFallback // update the subscription only if it was successful
+					s.logger.Info("Resubscribed to fallback new task subscription")
 				}
 			}
 		}
+		errorPairChannel <- ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
 	}()
 
-	return errorChannel, nil
+	return nil
 }
 
 func (s *AvsSubscriber) processNewBatchV2(batch *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, batchesSet map[[32]byte]struct{}, newBatchMutex *sync.Mutex, newTaskCreatedChan chan<- *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) {

core/chainio/retryable.go

@@ -2,6 +2,7 @@ package chainio
 
 import (
 	"context"
+	"github.com/rs/zerolog/log"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum"
@@ -206,6 +207,7 @@ func SubscribeToNewTasksV2Retryable(
 	config *retry.RetryParams,
 ) (event.Subscription, error) {
 	subscribe_func := func() (event.Subscription, error) {
+		log.Info().Msg("Subscribing to NewBatchV2")
 		return serviceManager.WatchNewBatchV2(opts, newTaskCreatedChan, batchMerkleRoot)
 	}
 	return retry.RetryWithData(subscribe_func, config)
@@ -225,6 +227,7 @@ func SubscribeToNewTasksV3Retryable(
 	config *retry.RetryParams,
 ) (event.Subscription, error) {
 	subscribe_func := func() (event.Subscription, error) {
+		log.Info().Msg("Subscribing to NewBatchV3")
 		return serviceManager.WatchNewBatchV3(opts, newTaskCreatedChan, batchMerkleRoot)
 	}
 	return retry.RetryWithData(subscribe_func, config)

network_params.yaml

@@ -1,7 +1,7 @@
 participants:
   - el_type: reth
     cl_type: lighthouse
-    count: 2
+    count: 3
     validator_count: 32
 
 ethereum_metrics_exporter_enabled: true