Refactor with retries

Author: Julian Ventura

aggregator/pkg/server.go

@@ -4,15 +4,14 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"net/http"
-	"net/rpc"
-	"strings"
-	"time"
-
 	"github.com/Layr-Labs/eigensdk-go/crypto/bls"
 	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
 	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/types"
+	"net/http"
+	"net/rpc"
+	"strings"
+	"time"
 )
 
 func (agg *Aggregator) ServeOperators() error {
@@ -37,53 +36,13 @@ func (agg *Aggregator) ServeOperators() error {
 	return err
 }
 
-// Waits for the arrival of task associated with signedTaskResponse and returns true on success or false on failure
-// If the task is not present in the internal map, it will try to fetch it from logs and retry.
-// The number of retries is specified by `waitForEventRetries`, and the waiting time between each by `waitForEventSleepSeconds`
-func (agg *Aggregator) waitForTaskAndFetchIfLost(signedTaskResponse *types.SignedTaskResponse) bool {
-	for i := 0; i < waitForEventRetries; i++ {
-		// Lock
-		agg.taskMutex.Lock()
-		agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Check if task is present")
-		_, ok := agg.batchesIdxByIdentifierHash[signedTaskResponse.BatchIdentifierHash]
-		// Unlock
-		agg.logger.Info("- Unlocked Resources: Check if task is present")
-		agg.taskMutex.Unlock()
-		if ok {
-			return true
-		}
-
-		// Task was not found in internal map, let's try to fetch it from logs
-		agg.logger.Info("Trying to fetch missed task from logs...")
-		batch, err := agg.avsReader.GetPendingBatchFromMerkleRoot(signedTaskResponse.BatchMerkleRoot)
-
-		if err == nil && batch != nil {
-			agg.logger.Info("Found missed task in logs with merkle root 0x%e", batch.BatchMerkleRoot)
-			// Adding new task will fail only if it already exists
-			agg.AddNewTask(batch.BatchMerkleRoot, batch.SenderAddress, batch.TaskCreatedBlock)
-			return true
-		}
-
-		if err != nil {
-			agg.logger.Warn("Error fetching task from logs: %v", err)
-		}
-
-		if batch == nil {
-			agg.logger.Info("Task not found in logs")
-		}
-
-		// Task was not found, wait and retry
-		time.Sleep(waitForEventSleepSeconds)
-	}
-
-	return false
-}
-
-// Aggregator Methods
+// ~~ AGGREGATOR METHODS ~~
 // This is the list of methods that the Aggregator exposes to the Operator
 // The Operator can call these methods to interact with the Aggregator
 // This methods are automatically registered by the RPC server
-// This takes a response an adds it to the internal. If reaching the quorum, it sends the aggregated signatures to ethereum
+
+// Takes a response from an operator and process it. After processing the response, the associated task may reach quorum, triggering a BLS service response.
+// If the task related to the response is not known to the aggregator (not stored in internal map), it will try to fetch it from the logs.
 // Returns:
 //   - 0: Success
 //   - 1: Error
@@ -97,21 +56,22 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
 	taskIndex, err := agg.GetTaskIndex(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")
-		*reply = 1
-		return nil
-	}
-
-	agg.taskMutex.Lock()
-	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Get task taskIndex")
-	taskIndex, ok := agg.batchesIdxByIdentifierHash[signedTaskResponse.BatchIdentifierHash]
-	// Unlock
-	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Get task taskIndex")
-	agg.taskMutex.Unlock()
-	if !ok {
-		agg.logger.Errorf("Unexpected error fetching for task with merkle root 0x%x", signedTaskResponse.BatchMerkleRoot)
-		*reply = 1
-		return nil
+		agg.logger.Warn("Task not found in the internal map, might have been missed. Trying to fetch it from logs")
+		batch, err := agg.avsReader.GetPendingBatchFromMerkleRoot(signedTaskResponse.BatchMerkleRoot)
+		if err != nil {
+			agg.logger.Warn("Pending task with merkle root 0x%x not found in logs")
+			*reply = 1
+			return nil
+		}
+		agg.logger.Info("Task was found in the logs, adding it to the internal map")
+		agg.AddNewTask(batch.BatchMerkleRoot, batch.SenderAddress, batch.TaskCreatedBlock)
+		taskIndex, err = agg.GetTaskIndex(signedTaskResponse.BatchIdentifierHash)
+		if err != nil {
+			// This shouldn't happen, since we just added the task
+			agg.logger.Error("Unexpected error trying to get taskIndex from internal map")
+			*reply = 1
+			return nil
+		}
 	}
 
 	agg.telemetry.LogOperatorResponse(signedTaskResponse.BatchMerkleRoot, signedTaskResponse.OperatorId)
@@ -194,11 +154,11 @@ func (agg *Aggregator) ProcessNewSignature(ctx context.Context, taskIndex uint32
 func (agg *Aggregator) GetTaskIndex(batchIdentifierHash [32]byte) (uint32, error) {
 	getTaskIndex_func := func() (uint32, error) {
 		agg.taskMutex.Lock()
-		agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Starting processing of Response")
+		agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Get task index")
 		taskIndex, ok := agg.batchesIdxByIdentifierHash[batchIdentifierHash]
+		agg.taskMutex.Unlock()
+		agg.logger.Info("- Unlocked Resources: Get task index")
 		if !ok {
-			agg.taskMutex.Unlock()
-			agg.logger.Info("- Unlocked Resources: Task not found in the internal map")
 			return taskIndex, fmt.Errorf("Task not found in the internal map")
 		} else {
 			return taskIndex, nil

core/chainio/avs_reader.go

@@ -158,21 +158,18 @@ func (r *AvsReader) GetOldTaskHash(nBlocksOld uint64, interval uint64) (*[32]byt
 // Returns a pending batch from its merkle root or nil if it doesn't exist
 // Searches the last `BatchFetchBlocksRange` blocks at most
 func (r *AvsReader) GetPendingBatchFromMerkleRoot(merkleRoot [32]byte) (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, error) {
-	latestBlock, err := r.AvsContractBindings.ethClient.BlockNumber(context.Background())
+	latestBlock, err := r.BlockNumberRetryable(context.Background())
 	if err != nil {
-		latestBlock, err = r.AvsContractBindings.ethClientFallback.BlockNumber(context.Background())
-		if err != nil {
-			return nil, fmt.Errorf("failed to get latest block number: %w", err)
-		}
+		return nil, fmt.Errorf("Failed to get latest block number: %w", err)
 	}
 
 	var fromBlock uint64 = 0
 
 	if latestBlock > BatchFetchBlocksRange {
-		fromBlock = latestBlock - BatchFetchBlocksRange
+		fromBlock = latestBlock - BatchFetchBlocksRange // TODO: Add this to config
 	}
 
-	logs, err := r.AvsContractBindings.ServiceManager.FilterNewBatchV3(&bind.FilterOpts{Start: fromBlock, End: &latestBlock, Context: context.Background()}, [][32]byte{merkleRoot})
+	logs, err := r.FilterBatchV3Retryable(&bind.FilterOpts{Start: fromBlock, End: &latestBlock, Context: context.Background()}, [][32]byte{merkleRoot})
 	if err != nil {
 		return nil, err
 	}
@@ -181,17 +178,19 @@ func (r *AvsReader) GetPendingBatchFromMerkleRoot(merkleRoot [32]byte) (*service
 	}
 
 	if !logs.Next() {
-		return nil, nil //not an error, but no tasks found
+		return nil, nil //Not an error, but no tasks found
 	}
 
 	batch := logs.Event
 
 	batchIdentifier := append(batch.BatchMerkleRoot[:], batch.SenderAddress[:]...)
 	batchIdentifierHash := *(*[32]byte)(crypto.Keccak256(batchIdentifier))
-	state, err := r.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.BatchesState(nil, batchIdentifierHash)
+	state, err := r.BatchesStateRetryable(nil, batchIdentifierHash)
+
 	if err != nil {
 		return nil, err
 	}
+
 	if state.Responded {
 		return nil, nil
 	}

core/chainio/retryable.go

@@ -202,3 +202,70 @@ func SubscribeToNewTasksV3Retryable(
 	}
 	return retry.RetryWithData(subscribe_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
 }
+
+// |---AVS_READER---|
+
+// TODO: These functions are being copied from AvsSubscriber and should be refactorized
+// we don't actually need access to the AvsReader, AvsSubscriber or AbsWriter, but instead to the AvsContractBindings
+
+// TODO: We should also add the fallback calls to the functions which are missing it
+
+/*
+- All errors are considered Transient Errors
+- Retry times (3 retries): 1 sec, 2 sec, 4 sec.
+*/
+func (r *AvsReader) BlockNumberRetryable(ctx context.Context) (uint64, error) {
+	latestBlock_func := func() (uint64, error) {
+		// Try with main connection
+		latestBlock, err := r.AvsContractBindings.ethClient.BlockNumber(ctx)
+		if err != nil {
+			// If error try with fallback connection
+			latestBlock, err = r.AvsContractBindings.ethClientFallback.BlockNumber(ctx)
+		}
+		return latestBlock, err
+	}
+	return retry.RetryWithData(latestBlock_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
+}
+
+/*
+- All errors are considered Transient Errors
+- Retry times (3 retries): 1 sec, 2 sec, 4 sec.
+*/
+func (r *AvsReader) FilterBatchV3Retryable(opts *bind.FilterOpts, batchMerkleRoot [][32]byte) (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3Iterator, error) {
+	filterNewBatchV2_func := func() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3Iterator, error) {
+		// Try with main connection
+		batch, err := r.AvsContractBindings.ServiceManager.FilterNewBatchV3(opts, batchMerkleRoot)
+		if err != nil {
+			// If error try with fallback connection
+			batch, err = r.AvsContractBindings.ServiceManagerFallback.FilterNewBatchV3(opts, batchMerkleRoot)
+		}
+		return batch, err
+	}
+	return retry.RetryWithData(filterNewBatchV2_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
+}
+
+/*
+- All errors are considered Transient Errors
+- Retry times (3 retries): 12 sec (1 Blocks), 24 sec (2 Blocks), 48 sec (4 Blocks)
+*/
+func (r *AvsReader) BatchesStateRetryable(opts *bind.CallOpts, arg0 [32]byte) (struct {
+	TaskCreatedBlock      uint32
+	Responded             bool
+	RespondToTaskFeeLimit *big.Int
+}, error) {
+	batchState_func := func() (struct {
+		TaskCreatedBlock      uint32
+		Responded             bool
+		RespondToTaskFeeLimit *big.Int
+	}, error) {
+		// Try with main connection
+		state, err := r.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.BatchesState(opts, arg0)
+		if err != nil {
+			// If error try with fallback connection
+			state, err = r.AvsContractBindings.ServiceManagerFallback.ContractAlignedLayerServiceManagerCaller.BatchesState(opts, arg0)
+		}
+		return state, err
+	}
+
+	return retry.RetryWithData(batchState_func, retry.MinDelayChain, retry.RetryFactor, retry.NumRetries, retry.MaxIntervalChain, retry.MaxElapsedTime)
+}