refactor retry logic for SendSignedTaskResponseToAggregator + Local testnet works w/ telemetry

Author: PatStiles

config-files/config-operator-test.yaml

@@ -0,0 +1,41 @@
+# Common variables for all the services
+# 'production' only prints info and above. 'development' also prints debug
+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:8545'
+eth_ws_url: 'ws://localhost:8545'
+eth_ws_url_fallback: 'ws://localhost:8545'
+eigen_metrics_ip_port_address: 'localhost:9090'
+
+## ECDSA Configurations
+ecdsa:
+  private_key_store_path: '../config-files/devnet/keys/operator-1.ecdsa.key.json'
+  private_key_store_password: ''
+
+## BLS Configurations
+bls:
+  private_key_store_path: '../config-files/devnet/keys/operator-1.bls.key.json'
+  private_key_store_password: ''
+
+## Operator Configurations
+operator:
+  aggregator_rpc_server_ip_port_address: localhost:8090
+  operator_tracker_ip_port_address: http://localhost:4001
+  address: 0x70997970C51812dc3A010C7d01b50e0d17dc79C8
+  earnings_receiver_address: 0x70997970C51812dc3A010C7d01b50e0d17dc79C8
+  delegation_approver_address: '0x0000000000000000000000000000000000000000'
+  staker_opt_out_window_blocks: 0
+  metadata_url: 'https://yetanotherco.github.io/operator_metadata/metadata.json'
+  enable_metrics: true
+  metrics_ip_port_address: localhost:9092
+  max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: '../config-files/operator-1.last_processed_batch.json'
+
+# Operators variables needed for register it in EigenLayer
+el_delegation_manager_address: '0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9'
+private_key_store_path: config-files/devnet/keys/operator-1.ecdsa.key.json
+bls_private_key_store_path: config-files/devnet/keys/operator-1.bls.key.json
+signer_type: local_keystore
+chain_id: 31337

operator/pkg/operator.go

@@ -334,7 +334,7 @@ func (o *Operator) handleNewBatchLogV2(newBatchLog *servicemanager.ContractAlign
 		hex.EncodeToString(signedTaskResponse.SenderAddress[:]),
 	)
 
-	o.aggRpcClient.SendSignedTaskResponseToAggregator(&signedTaskResponse)
+	o.aggRpcClient.SendSignedTaskResponseToAggregatorRetryable(&signedTaskResponse)
 }
 func (o *Operator) ProcessNewBatchLogV2(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) error {
 
@@ -415,7 +415,7 @@ func (o *Operator) handleNewBatchLogV3(newBatchLog *servicemanager.ContractAlign
 		hex.EncodeToString(signedTaskResponse.SenderAddress[:]),
 	)
 
-	o.aggRpcClient.SendSignedTaskResponseToAggregator(&signedTaskResponse)
+	o.aggRpcClient.SendSignedTaskResponseToAggregatorRetryable(&signedTaskResponse)
 }
 func (o *Operator) ProcessNewBatchLogV3(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3) error {
 

operator/pkg/rpc_client.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/Layr-Labs/eigensdk-go/logging"
+	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/types"
 )
 
@@ -16,11 +17,6 @@ type AggregatorRpcClient struct {
 	logger               logging.Logger
 }
 
-const (
-	MaxRetries    = 10
-	RetryInterval = 10 * time.Second
-)
-
 func NewAggregatorRpcClient(aggregatorIpPortAddr string, logger logging.Logger) (*AggregatorRpcClient, error) {
 	client, err := rpc.DialHTTP("tcp", aggregatorIpPortAddr)
 	if err != nil {
@@ -34,31 +30,29 @@ func NewAggregatorRpcClient(aggregatorIpPortAddr string, logger logging.Logger)
 	}, nil
 }
 
-// SendSignedTaskResponseToAggregator is the method called by operators via RPC to send
-// their signed task response.
-func (c *AggregatorRpcClient) SendSignedTaskResponseToAggregator(signedTaskResponse *types.SignedTaskResponse) {
-	var reply uint8
-	for retries := 0; retries < MaxRetries; retries++ {
+func SendSignedTaskResponse(c *AggregatorRpcClient, signedTaskResponse *types.SignedTaskResponse) func() (uint8, error) {
+	send_task_func := func() (uint8, error) {
+		var reply uint8
 		err := c.rpcClient.Call("Aggregator.ProcessOperatorSignedTaskResponseV2", signedTaskResponse, &reply)
 		if err != nil {
 			c.logger.Error("Received error from aggregator", "err", err)
 			if errors.Is(err, rpc.ErrShutdown) {
 				c.logger.Error("Aggregator is shutdown. Reconnecting...")
-				client, err := rpc.DialHTTP("tcp", c.aggregatorIpPortAddr)
-				if err != nil {
-					c.logger.Error("Could not reconnect to aggregator", "err", err)
-					time.Sleep(RetryInterval)
-				} else {
-					c.rpcClient = client
-					c.logger.Info("Reconnected to aggregator")
-				}
-			} else {
-				c.logger.Infof("Received error from aggregator: %s. Retrying ProcessOperatorSignedTaskResponseV2 RPC call...", err)
-				time.Sleep(RetryInterval)
 			}
 		} else {
 			c.logger.Info("Signed task response header accepted by aggregator.", "reply", reply)
-			return
 		}
+		return reply, err
 	}
+	return send_task_func
+}
+
+// SendSignedTaskResponseToAggregator is the method called by operators via RPC to send
+// their signed task response.
+func (c *AggregatorRpcClient) SendSignedTaskResponseToAggregatorRetryable(signedTaskResponse *types.SignedTaskResponse) (uint8, error) {
+	config := retry.DefaultRetryConfig()
+	config.NumRetries = 10
+	config.Multiplier = 1 // Constant retry interval
+	config.InitialInterval = 10 * time.Second
+	return retry.RetryWithData(SendSignedTaskResponse(c, signedTaskResponse), config)
 }