fix: send aggregated response bump

aggregator/pkg/aggregator.go

@@ -225,6 +225,13 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 const MaxSentTxRetries = 5
 
 func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsAggregationServiceResponse) {
+	defer func() {
+ err := recover() //stops panics
+ if err != nil {
+ agg.logger.Error("handleBlsAggServiceResponse recovered from panic", "err", err)
+ }
+	}()
+
 	agg.taskMutex.Lock()
 	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Fetching task data")
 	batchIdentifierHash := agg.batchesIdentifierHashByIdx[blsAggServiceResp.TaskIndex]
@@ -277,10 +284,15 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 	}
 
 	agg.logger.Info("Sending aggregated response onchain", "taskIndex", blsAggServiceResp.TaskIndex,
- "batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
+ "batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]), "merkleRoot", "0x"+hex.EncodeToString(batchData.BatchMerkleRoot[:]))
 	receipt, err := agg.sendAggregatedResponse(batchIdentifierHash, batchData.BatchMerkleRoot, batchData.SenderAddress, nonSignerStakesAndSignature)
 	if err == nil {
- agg.telemetry.TaskSentToEthereum(batchData.BatchMerkleRoot, receipt.TxHash.String())
+ // In some cases, we may fail to retrieve the receipt for the transaction.
+ txHash := "Unknown"
+ if receipt != nil {
+ txHash = receipt.TxHash.String()
+ }
+ agg.telemetry.TaskSentToEthereum(batchData.BatchMerkleRoot, txHash)
  agg.logger.Info("Aggregator successfully responded to task",
  "taskIndex", blsAggServiceResp.TaskIndex,
  "batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
@@ -428,7 +440,7 @@ func (agg *Aggregator) ClearTasksFromMaps() {
 	defer func() {
  err := recover() //stops panics
  if err != nil {
- agg.logger.Error("Recovered from panic", "err", err)
+ agg.logger.Error("ClearTasksFromMaps Recovered from panic", "err", err)
  }
 	}()
 

config-files/config-aggregator.yaml

@@ -38,9 +38,10 @@ 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)
- gas_base_bump_percentage: 10 # How much to bump gas price when responding to task. Suggested value 10%
- gas_bump_incremental_percentage: 2 # An extra percentage to bump every retry i*2 when responding to task. Suggested value 2%
- time_to_wait_before_bump: 36s # The time to wait for the receipt when responding to task. Suggested value 36 seconds (3 blocks)
+ gas_base_bump_percentage: 25 # Percentage to overestimate gas price when sending a task
+ gas_bump_incremental_percentage: 20 # An extra percentage to overestimate in each bump of respond to task. This is additive between tries
+ # Gas used formula = est_gas_by_node * (gas_base_bump_percentage + gas_bum_incremental_percentage * i) / 100, where i is the iteration number.
+ time_to_wait_before_bump: 72s # The time to wait for the receipt when responding to task. Suggested value 72 seconds (6 blocks)
 
 ## Operator Configurations
 # operator:

core/chainio/avs_writer.go

@@ -2,6 +2,7 @@ package chainio
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
 	"math/big"
 	"time"
@@ -75,9 +76,19 @@ func NewAvsWriterFromConfig(baseConfig *config.BaseConfig, ecdsaConfig *config.E
 	}, nil
 }
 
-// Sends AggregatedResponse and waits for the receipt for three blocks, if not received
-// it will try again bumping the last tx gas price based on `CalculateGasPriceBump`
-// This process happens indefinitely until the transaction is included.
+// SendAggregatedResponse continuously sends a RespondToTask transaction until it is included in the blockchain.
+// This function:
+// 1. Simulates the transaction to calculate the nonce and initial gas price without broadcasting it.
+// 2. Repeatedly attempts to send the transaction, bumping the gas price after `timeToWaitBeforeBump` has passed.
+// 3. Monitors for the receipt of previously sent transactions or checks the state to confirm if the response
+// has already been processed (e.g., by another transaction).
+// 4. Validates that the aggregator and batcher have sufficient balance to cover transaction costs before sending.
+//
+// Returns:
+// - A transaction receipt if the transaction is successfully included in the blockchain.
+// - If no receipt is found, but the batch state indicates the response has already been processed, it exits
+// without an error (returning `nil, nil`).
+// - An error if the process encounters a fatal issue (e.g., permanent failure in verifying balances or state).
 func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature, gasBumpPercentage uint, gasBumpIncrementalPercentage uint, timeToWaitBeforeBump time.Duration, onGasPriceBumped func(*big.Int)) (*types.Receipt, error) {
 	txOpts := *w.Signer.GetTxOpts()
 	txOpts.NoSend = true // simulate the transaction
@@ -93,39 +104,73 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
 	txOpts.NoSend = false
 	i := 0
 
+	var sentTxs []*types.Transaction
+
+	batchMerkleRootHashString := hex.EncodeToString(batchMerkleRoot[:])
+
 	respondToTaskV2Func := func() (*types.Receipt, error) {
  gasPrice, err := utils.GetGasPriceRetryable(w.Client, w.ClientFallback)
  if err != nil {
  return nil, err
  }
-
- bumpedGasPrice := utils.CalculateGasPriceBumpBasedOnRetry(gasPrice, gasBumpPercentage, gasBumpIncrementalPercentage, i)
- // new bumped gas price must be higher than the last one (this should hardly ever happen though)
- if bumpedGasPrice.Cmp(txOpts.GasPrice) > 0 {
- txOpts.GasPrice = bumpedGasPrice
+ previousTxGasPrice := txOpts.GasPrice
+ // in order to avoid replacement transaction underpriced
+ // the bumped gas price has to be at least 10% higher than the previous one.
+ minimumGasPriceBump := utils.CalculateGasPriceBumpBasedOnRetry(previousTxGasPrice, 10, 0, 0)
+ suggestedBumpedGasPrice := utils.CalculateGasPriceBumpBasedOnRetry(
+ gasPrice,
+ gasBumpPercentage,
+ gasBumpIncrementalPercentage,
+ i,
+ )
+ // check the new gas price is sufficiently bumped.
+ // if the suggested bump does not meet the minimum threshold, use a fallback calculation to slightly increment the previous gas price.
+ if suggestedBumpedGasPrice.Cmp(minimumGasPriceBump) > 0 {
+ txOpts.GasPrice = suggestedBumpedGasPrice
  } else {
- // bump the last tx gas price a little by `gasBumpIncrementalPercentage` to replace it.
- txOpts.GasPrice = utils.CalculateGasPriceBumpBasedOnRetry(txOpts.GasPrice, gasBumpIncrementalPercentage, 0, 0)
+ txOpts.GasPrice = minimumGasPriceBump
  }
 
  if i > 0 {
+ w.logger.Infof("Trying to get old sent transaction receipt before sending a new transaction", "merkle root", batchMerkleRootHashString)
+ for _, tx := range sentTxs {
+ receipt, _ := w.Client.TransactionReceipt(context.Background(), tx.Hash())
+ if receipt == nil {
+ receipt, _ = w.ClientFallback.TransactionReceipt(context.Background(), tx.Hash())
+ if receipt != nil {
+ w.checkIfAggregatorHadToPaidForBatcher(tx, batchIdentifierHash)
+ return receipt, nil
+ }
+ }
+ }
+ w.logger.Infof("Receipts for old transactions not found, will check if the batch state has been responded", "merkle root", batchMerkleRootHashString)
+ batchState, _ := w.BatchesStateRetryable(&bind.CallOpts{}, batchIdentifierHash)
+ if batchState.Responded {
+ w.logger.Infof("Batch state has been already responded", "merkle root", batchMerkleRootHashString)
+ return nil, nil
+ }
+ w.logger.Infof("Batch state has not been responded yet, will send a new tx", "merkle root", batchMerkleRootHashString)
+
  onGasPriceBumped(txOpts.GasPrice)
  }
 
  // We compare both Aggregator funds and Batcher balance in Aligned against respondToTaskFeeLimit
  // Both are required to have some balance, more details inside the function
  err = w.checkAggAndBatcherHaveEnoughBalance(simTx, txOpts, batchIdentifierHash, senderAddress)
  if err != nil {
+ w.logger.Errorf("Permanent error when checking aggregator and batcher balances, err %v", err, "merkle root", batchMerkleRootHashString)
  return nil, retry.PermanentError{Inner: err}
  }
 
- w.logger.Infof("Sending RespondToTask transaction with a gas price of %v", txOpts.GasPrice)
-
+ w.logger.Infof("Sending RespondToTask transaction with a gas price of %v", txOpts.GasPrice, "merkle root", batchMerkleRootHashString)
  realTx, err := w.RespondToTaskV2Retryable(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
  if err != nil {
+ w.logger.Errorf("Respond to task transaction err, %v", err, "merkle root", batchMerkleRootHashString)
  return nil, err
  }
+ 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)
  if receipt != nil {
  w.checkIfAggregatorHadToPaidForBatcher(realTx, batchIdentifierHash)
@@ -136,14 +181,18 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
  // we increment the i here to add an incremental percentage to increase the odds of being included in the next blocks
  i++
 
- w.logger.Infof("RespondToTask receipt waiting timeout has passed, will try again...")
+ w.logger.Infof("RespondToTask receipt waiting timeout has passed, will try again...", "merkle_root", batchMerkleRootHashString)
  if err != nil {
  return nil, err
  }
  return nil, fmt.Errorf("transaction failed")
 	}
 
-	return retry.RetryWithData(respondToTaskV2Func, retry.MinDelay, retry.RetryFactor, 0, retry.MaxInterval, 0)
+	// 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)
 }
 
 // Calculates the transaction cost from the receipt and compares it with the batcher respondToTaskFeeLimit
@@ -171,7 +220,9 @@ func (w *AvsWriter) checkIfAggregatorHadToPaidForBatcher(tx *types.Transaction,
 func (w *AvsWriter) checkAggAndBatcherHaveEnoughBalance(tx *types.Transaction, txOpts bind.TransactOpts, batchIdentifierHash [32]byte, senderAddress [20]byte) error {
 	w.logger.Info("Checking if aggregator and batcher have enough balance for the transaction")
 	aggregatorAddress := txOpts.From
-	txCost := new(big.Int).Mul(new(big.Int).SetUint64(tx.Gas()), txOpts.GasPrice)
+	txGasAsBigInt := new(big.Int).SetUint64(tx.Gas())
+	txGasPrice := txOpts.GasPrice
+	txCost := new(big.Int).Mul(txGasAsBigInt, txGasPrice)
 	w.logger.Info("Transaction cost", "cost", txCost)
 
 	batchState, err := w.BatchesStateRetryable(&bind.CallOpts{}, batchIdentifierHash)
@@ -183,8 +234,8 @@ func (w *AvsWriter) checkAggAndBatcherHaveEnoughBalance(tx *types.Transaction, t
 	respondToTaskFeeLimit := batchState.RespondToTaskFeeLimit
 	w.logger.Info("Checking balance against Batch RespondToTaskFeeLimit", "RespondToTaskFeeLimit", respondToTaskFeeLimit)
 	// Note: we compare both Aggregator funds and Batcher balance in Aligned against respondToTaskFeeLimit
-// Batcher will pay up to respondToTaskFeeLimit, for this he needs that amount of funds in Aligned
-// Aggregator will pay any extra cost, for this he needs at least respondToTaskFeeLimit in his balance
+	// Batcher will pay up to respondToTaskFeeLimit, for this he needs that amount of funds in Aligned
+	// Aggregator will pay any extra cost, for this he needs at least respondToTaskFeeLimit in his balance
 	return w.compareBalances(respondToTaskFeeLimit, aggregatorAddress, senderAddress)
 }
 

core/chainio/retryable.go

@@ -30,6 +30,7 @@ func (w *AvsWriter) RespondToTaskV2Retryable(opts *bind.TransactOpts, batchMerkl
  // If error try with fallback
  tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(opts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
  }
+
  return tx, err
 	}
 	return retry.RetryWithData(respondToTaskV2_func, retry.MinDelayChain, retry.RetryFactor, retry.NumRetries, retry.MaxIntervalChain, retry.MaxElapsedTime)

core/retry.go

@@ -104,7 +104,7 @@ func RetryWithData[T any](functionToRetry func() (T, error), minDelay time.Durat
  if panic_err, ok := r.(error); ok {
  err = panic_err
  } else {
- err = fmt.Errorf("panicked: %v", panic_err)
+ err = fmt.Errorf("RetryWithData panicked: %v", panic_err)
  }
  }
  }()
@@ -151,7 +151,7 @@ func Retry(functionToRetry func() error, minDelay time.Duration, factor float64,
  if panic_err, ok := r.(error); ok {
  err = panic_err
  } else {
- err = fmt.Errorf("panicked: %v", panic_err)
+ err = fmt.Errorf("Retry panicked: %v", panic_err)
  }
  }
  }()