Add block fetch range to config, fix related bug

Author: Julian Ventura

aggregator/pkg/server.go

@@ -57,9 +57,9 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
 
 	if err != 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")
+		batch, err := agg.avsReader.GetPendingBatchFromMerkleRoot(signedTaskResponse.BatchMerkleRoot, agg.AggregatorConfig.Aggregator.PendingBatchFetchBlockRange)
+		if err != nil || batch == nil {
+			agg.logger.Warnf("Pending task with merkle root 0x%x not found in logs", signedTaskResponse.BatchMerkleRoot)
 			*reply = 1
 			return nil
 		}

config-files/config-aggregator.yaml

@@ -38,6 +38,7 @@ aggregator:
   garbage_collector_tasks_age: 20 #The age of tasks that will be removed by the GC, in blocks. Suggested value for prod: '216000' (30 days)
   garbage_collector_tasks_interval: 10 #The interval of queried blocks to get an old batch. Suggested value for prod: '900' (3 hours)
   bls_service_task_timeout: 168h # The timeout of bls aggregation service tasks. Suggested value for prod '168h' (7 days)
+  pending_batch_fetch_block_range: 1000 #The interval of queried blocks to get a pending batch by logs. Suggested valued for prod `1000`
 
 ## Operator Configurations
 # operator:

core/chainio/avs_reader.go

@@ -17,10 +17,6 @@ import (
 	"github.com/Layr-Labs/eigensdk-go/logging"
 )
 
-const (
-	BatchFetchBlocksRange uint64 = 1000
-)
-
 type AvsReader struct {
 	*sdkavsregistry.ChainReader
 	AvsContractBindings            *AvsServiceBindings
@@ -156,17 +152,17 @@ 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) {
+// Searches the last `blockRange` blocks at most
+func (r *AvsReader) GetPendingBatchFromMerkleRoot(merkleRoot [32]byte, blockRange uint64) (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, error) {
 	latestBlock, err := r.BlockNumberRetryable(context.Background())
 	if err != nil {
 		return nil, fmt.Errorf("Failed to get latest block number: %w", err)
 	}
 
 	var fromBlock uint64 = 0
 
-	if latestBlock > BatchFetchBlocksRange {
-		fromBlock = latestBlock - BatchFetchBlocksRange // TODO: Add this to config
+	if latestBlock > blockRange {
+		fromBlock = latestBlock - blockRange
 	}
 
 	logs, err := r.FilterBatchV3Retryable(&bind.FilterOpts{Start: fromBlock, End: &latestBlock, Context: context.Background()}, [][32]byte{merkleRoot})

core/config/aggregator.go

@@ -25,6 +25,7 @@ type AggregatorConfig struct {
 		GarbageCollectorTasksAge      uint64
 		GarbageCollectorTasksInterval uint64
 		BlsServiceTaskTimeout         time.Duration
+		PendingBatchFetchBlockRange   uint64
 	}
 }
 
@@ -40,6 +41,7 @@ type AggregatorConfigFromYaml struct {
 		GarbageCollectorTasksAge      uint64         `yaml:"garbage_collector_tasks_age"`
 		GarbageCollectorTasksInterval uint64         `yaml:"garbage_collector_tasks_interval"`
 		BlsServiceTaskTimeout         time.Duration  `yaml:"bls_service_task_timeout"`
+		PendingBatchFetchBlockRange   uint64         `yaml:"pending_batch_fetch_block_range"`
 	} `yaml:"aggregator"`
 }
 
@@ -85,6 +87,7 @@ func NewAggregatorConfig(configFilePath string) *AggregatorConfig {
 			GarbageCollectorTasksAge      uint64
 			GarbageCollectorTasksInterval uint64
 			BlsServiceTaskTimeout         time.Duration
+			PendingBatchFetchBlockRange   uint64
 		}(aggregatorConfigFromYaml.Aggregator),
 	}
 }