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

core/retry_test.go

@@ -14,18 +14,22 @@ import (
 	"time"
 
 	"github.com/Layr-Labs/eigensdk-go/chainio/clients/eth"
+	"github.com/Layr-Labs/eigensdk-go/crypto/bls"
 	rpccalls "github.com/Layr-Labs/eigensdk-go/metrics/collectors/rpc_calls"
 	backoff "github.com/cenkalti/backoff/v4"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
+	aggregator "github.com/yetanotherco/aligned_layer/aggregator/pkg"
 	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
 	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/chainio"
 	"github.com/yetanotherco/aligned_layer/core/config"
+	core_types "github.com/yetanotherco/aligned_layer/core/types"
 	"github.com/yetanotherco/aligned_layer/core/utils"
+	operator "github.com/yetanotherco/aligned_layer/operator/pkg"
 )
 
 func DummyFunction(x uint64) (uint64, error) {
@@ -923,3 +927,69 @@ func TestBatchersBalances(t *testing.T) {
 		return
 	}
 }
+
+func TestSendSignedTaskResponse(t *testing.T) {
+	cmd, _, err := SetupAnvil(8545)
+	if err != nil {
+		t.Errorf("Error setting up Anvil: %s\n", err)
+	}
+
+	aggregatorConfig := config.NewAggregatorConfig("../config-files/config-aggregator-test.yaml")
+	operatorConfig := config.NewOperatorConfig("../config-files/config-operator-test.yaml")
+
+	agg, err := aggregator.NewAggregator(*aggregatorConfig)
+	if err != nil {
+		fmt.Errorf("Could not load aggregator %s", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		err = agg.Start(ctx)
+		if err != nil {
+			fmt.Errorf("Could not start Aggregator %s", err)
+		}
+	}()
+	if err != nil {
+		fmt.Errorf("Could not start Aggregator %s", err)
+	}
+
+	logger := operatorConfig.BaseConfig.Logger
+	rpcClient, err := operator.NewAggregatorRpcClient(operatorConfig.Operator.AggregatorServerIpPortAddress, logger)
+	// This is a mock to verify that a lost connection to the aggregator does not return anything.
+	disconnected_rpcClient, err := operator.NewAggregatorRpcClient("localhost:8999", logger)
+	if err != nil {
+		fmt.Errorf("Could not create RPC client: %s. Is aggregator running?", err)
+	}
+
+	//zero_bytes := [32]byte{}
+	signedTaskResponse := &core_types.SignedTaskResponse{
+		BatchIdentifierHash: [32]byte{},
+		BatchMerkleRoot:     [32]byte{},
+		SenderAddress:       [20]byte{},
+		BlsSignature:        bls.Signature{},
+		OperatorId:          [32]byte{},
+	}
+
+	task_resp_func := operator.SendSignedTaskResponse(rpcClient, signedTaskResponse)
+	_, err = task_resp_func()
+	assert.Nil(t, err)
+
+	fmt.Printf("stuff")
+	task_resp_func = operator.SendSignedTaskResponse(disconnected_rpcClient, signedTaskResponse)
+	// NOTE: This the expected behavior of a net/rpc/clint/Go for an invalid connection ref: https://cs.opensource.google/go/go/+/refs/tags/go1.23.3:src/net/rpc/client.go;l=298
+	assert.Panics(t, func() { _, err = task_resp_func() })
+	fmt.Printf("error: %s", err)
+	assert.Nil(t, err)
+
+	task_resp_func = operator.SendSignedTaskResponse(rpcClient, signedTaskResponse)
+	_, err = task_resp_func()
+	assert.Nil(t, err)
+
+	// Kill Aggregator
+	cancel()
+
+	if err := cmd.Process.Kill(); err != nil {
+		t.Errorf("Error killing process: %v\n", err)
+		return
+	}
+
+}

operator/pkg/operator.go

@@ -349,7 +349,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 {
 
@@ -430,7 +430,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)
 }