feat(aggregator): Aggregator Retry improvements

aggregator/pkg/aggregator.go

@@ -5,14 +5,12 @@ import (
 	"encoding/hex"
 	"fmt"
 	"math/big"
-	"strings"
 	"sync"
 	"time"
 
 	gethtypes "github.com/ethereum/go-ethereum/core/types"
 
 	"github.com/prometheus/client_golang/prometheus"
-	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/metrics"
 
 	sdkclients "github.com/Layr-Labs/eigensdk-go/chainio/clients"
@@ -396,7 +394,7 @@ func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]by
 	quorumNums := eigentypes.QuorumNums{eigentypes.QuorumNum(QUORUM_NUMBER)}
 	quorumThresholdPercentages := eigentypes.QuorumThresholdPercentages{eigentypes.QuorumThresholdPercentage(QUORUM_THRESHOLD)}
 
-	err := agg.InitializeNewTaskRetryable(batchIndex, taskCreatedBlock, quorumNums, quorumThresholdPercentages, agg.AggregatorConfig.Aggregator.BlsServiceTaskTimeout)
+	err := agg.blsAggregationService.InitializeNewTask(batchIndex, taskCreatedBlock, quorumNums, quorumThresholdPercentages, agg.AggregatorConfig.Aggregator.BlsServiceTaskTimeout)
 	if err != nil {
 		agg.logger.Fatalf("BLS aggregation service error when initializing new task: %s", err)
 	}
@@ -409,30 +407,6 @@ func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]by
 
 // |---RETRYABLE---|
 
-/*
-InitializeNewTaskRetryable
-Initialize a new task in the BLS Aggregation service
-  - Errors:
-    Permanent:
-  - TaskAlreadyInitializedError (Permanent): Task is already intialized in the BLS Aggregation service (https://github.com/Layr-Labs/eigensdk-go/blob/dev/services/bls_aggregation/blsagg.go#L27).
-    Transient:
-  - All others.
-  - Retry times (3 retries): 1 sec, 2 sec, 4 sec
-*/
-func (agg *Aggregator) InitializeNewTaskRetryable(batchIndex uint32, taskCreatedBlock uint32, quorumNums eigentypes.QuorumNums, quorumThresholdPercentages eigentypes.QuorumThresholdPercentages, timeToExpiry time.Duration) error {
-	initializeNewTask_func := func() error {
-		err := agg.blsAggregationService.InitializeNewTask(batchIndex, taskCreatedBlock, quorumNums, quorumThresholdPercentages, timeToExpiry)
-		if err != nil {
-			// Task is already initialized
-			if strings.Contains(err.Error(), "already initialized") {
-				err = retry.PermanentError{Inner: err}
-			}
-		}
-		return err
-	}
-	return retry.Retry(initializeNewTask_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
-}
-
 // Long-lived goroutine that periodically checks and removes old Tasks from stored Maps
 // It runs every GarbageCollectorPeriod and removes all tasks older than GarbageCollectorTasksAge
 // This was added because each task occupies memory in the maps, and we need to free it to avoid a memory leak

aggregator/pkg/server.go

@@ -50,7 +50,7 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
 		"operatorId", hex.EncodeToString(signedTaskResponse.OperatorId[:]))
 	taskIndex := uint32(0)
 
-	taskIndex, err := agg.GetTaskIndex(signedTaskResponse.BatchIdentifierHash)
+	taskIndex, err := agg.GetTaskIndexRetryable(signedTaskResponse.BatchIdentifierHash)
 
 	if err != nil {
 		agg.logger.Warn("Task not found in the internal map, operator signature will be lost. Batch may not reach quorum")
@@ -106,17 +106,25 @@ func (agg *Aggregator) ServerRunning(_ *struct{}, reply *int64) error {
 	return nil
 }
 
-func (agg *Aggregator) GetTaskIndex(batchIdentifierHash [32]byte) (uint32, error) {
+func GetTaskIndexFunc(agg *Aggregator, batchIdentifierHash [32]byte) func() (uint32, error) {
 	getTaskIndex_func := func() (uint32, error) {
 		agg.taskMutex.Lock()
 		taskIndex, ok := agg.batchesIdxByIdentifierHash[batchIdentifierHash]
 		agg.taskMutex.Unlock()
 		if !ok {
-			return taskIndex, fmt.Errorf("Task not found in the internal map")
+			return taskIndex, fmt.Errorf("task not found in the internal map")
 		} else {
 			return taskIndex, nil
 		}
 	}
+	return getTaskIndex_func
+}
 
-	return retry.RetryWithData(getTaskIndex_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
+// Checks Internal mapping for Signed Task Response, returns its TaskIndex.
+/*
+- All errors are considered Transient Errors
+- Retry times (3 retries): 1 sec, 2 sec, 4 sec
+*/
+func (agg *Aggregator) GetTaskIndexRetryable(batchIdentifierHash [32]byte) (uint32, error) {
+	return retry.RetryWithData(GetTaskIndexFunc(agg, batchIdentifierHash), retry.EthCallRetryConfig())
 }

core/chainio/avs_subscriber.go

@@ -68,13 +68,13 @@ func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicema
 	// Subscribe to new tasks
 	sub, err := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil)
 	if err != nil {
-		s.logger.Error("Primary failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.NumRetries, "err", err)
+		s.logger.Error("Primary failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.EthCallNumRetries, "err", err)
 		return nil, err
 	}
 
 	subFallback, err := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil)
 	if err != nil {
-		s.logger.Error("Fallback failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.NumRetries, "err", err)
+		s.logger.Error("Fallback failed to subscribe to new AlignedLayer V2 tasks after %d retries", retry.EthCallNumRetries, "err", err)
 		return nil, err
 	}
 	s.logger.Info("Subscribed to new AlignedLayer V2 tasks")

core/chainio/avs_writer.go

@@ -22,6 +22,14 @@ import (
 	"github.com/yetanotherco/aligned_layer/metrics"
 )
 
+const (
+	waitForTxMaxInterval                 = 2 * time.Second
+	waitForTxNumRetries                  = 0
+	respondToTaskV2NumRetries     uint64 = 0
+	respondToTaskV2MaxInterval           = time.Millisecond * 500
+	respondToTaskV2MaxElapsedTime        = 0
+)
+
 type AvsWriter struct {
 	*avsregistry.ChainWriter
 	AvsContractBindings *AvsServiceBindings
@@ -104,6 +112,18 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
 	txOpts.NoSend = false
 	i := 0
 
+	// Set Retry config for RespondToTaskV2
+	respondToTaskV2Config := retry.EthCallRetryConfig()
+	respondToTaskV2Config.NumRetries = respondToTaskV2NumRetries
+	respondToTaskV2Config.MaxInterval = respondToTaskV2MaxInterval
+	respondToTaskV2Config.MaxElapsedTime = respondToTaskV2MaxElapsedTime
+
+	// Set Retry config for WaitForTxRetryable
+	waitForTxConfig := retry.EthCallRetryConfig()
+	waitForTxConfig.MaxInterval = waitForTxMaxInterval
+	waitForTxConfig.NumRetries = waitForTxNumRetries
+	waitForTxConfig.MaxElapsedTime = timeToWaitBeforeBump
+
 	var sentTxs []*types.Transaction
 
 	batchMerkleRootHashString := hex.EncodeToString(batchMerkleRoot[:])
@@ -171,7 +191,8 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
 		sentTxs = append(sentTxs, realTx)
 
 		w.logger.Infof("Transaction sent, waiting for receipt", "merkle root", batchMerkleRootHashString)
-		receipt, err := utils.WaitForTransactionReceiptRetryable(w.Client, w.ClientFallback, realTx.Hash(), timeToWaitBeforeBump)
+		receipt, err := utils.WaitForTransactionReceiptRetryable(w.Client, w.ClientFallback, realTx.Hash(), waitForTxConfig)
+
 		if receipt != nil {
 			w.checkIfAggregatorHadToPaidForBatcher(realTx, batchIdentifierHash)
 			return receipt, nil
@@ -191,8 +212,7 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
 	// This just retries the bump of a fee in case of a timeout
 	// The wait is done before on WaitForTransactionReceiptRetryable, and all the functions are retriable,
 	// so this retry doesn't need to wait more time
-	maxInterval := time.Millisecond * 500
-	return retry.RetryWithData(respondToTaskV2Func, retry.MinDelay, retry.RetryFactor, 0, maxInterval, 0)
+	return retry.RetryWithData(respondToTaskV2Func, respondToTaskV2Config)
 }
 
 // Calculates the transaction cost from the receipt and compares it with the batcher respondToTaskFeeLimit