feat: aggregator bump fee if transaction was not included

.github/workflows/build-and-test-go.yml

@@ -24,6 +24,8 @@ jobs:
  with:
  go-version: "1.23"
  cache: false
+ - name: foundry-toolchain
+ uses: foundry-rs/foundry-toolchain@v1.2.0
  - name: Build SP1 bindings
  run: make build_sp1_linux
  - name: Build Old SP1 bindings

aggregator/pkg/aggregator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"math/big"
 	"strings"
 	"sync"
 	"time"
@@ -300,7 +301,21 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
  "senderAddress", hex.EncodeToString(senderAddress[:]),
  "batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
 
-	txHash, err := agg.avsWriter.SendAggregatedResponse(batchIdentifierHash, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	// This function is a callback that is called when the gas price is bumped on the avsWriter.SendAggregatedResponse
+	onGasPriceBumped := func(bumpedGasPrice *big.Int) {
+ agg.metrics.IncBumpedGasPriceForAggregatedResponse()
+ agg.telemetry.BumpedTaskGasPrice(batchMerkleRoot, bumpedGasPrice.String())
+	}
+	receipt, err := agg.avsWriter.SendAggregatedResponse(
+ batchIdentifierHash,
+ batchMerkleRoot,
+ senderAddress,
+ nonSignerStakesAndSignature,
+ agg.AggregatorConfig.Aggregator.GasBaseBumpPercentage,
+ agg.AggregatorConfig.Aggregator.GasBumpIncrementalPercentage,
+ agg.AggregatorConfig.Aggregator.TimeToWaitBeforeBump,
+ onGasPriceBumped,
+	)
 	if err != nil {
  agg.walletMutex.Unlock()
  agg.logger.Infof("- Unlocked Wallet Resources: Error sending aggregated response for batch %s. Error: %s", hex.EncodeToString(batchIdentifierHash[:]), err)
@@ -311,13 +326,6 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 	agg.walletMutex.Unlock()
 	agg.logger.Infof("- Unlocked Wallet Resources: Sending aggregated response for batch %s", hex.EncodeToString(batchIdentifierHash[:]))
 
-	receipt, err := utils.WaitForTransactionReceipt(
- agg.AggregatorConfig.BaseConfig.EthRpcClient, context.Background(), *txHash)
-	if err != nil {
- agg.telemetry.LogTaskError(batchMerkleRoot, err)
- return nil, err
-	}
-
 	agg.metrics.IncAggregatedResponses()
 
 	return receipt, nil

aggregator/pkg/telemetry.go

@@ -30,6 +30,11 @@ type TaskErrorMessage struct {
 	TaskError string `json:"error"`
 }
 
+type TaskGasPriceBumpMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+	BumpedGasPrice string `json:"bumped_gas_price"`
+}
+
 type TaskSentToEthereumMessage struct {
 	MerkleRoot string `json:"merkle_root"`
 	TxHash string `json:"tx_hash"`
@@ -96,6 +101,16 @@ func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error) {
 	}
 }
 
+func (t *Telemetry) BumpedTaskGasPrice(batchMerkleRoot [32]byte, bumpedGasPrice string) {
+	body := TaskGasPriceBumpMessage{
+ MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+ BumpedGasPrice: bumpedGasPrice,
+	}
+	if err := t.sendTelemetryMessage("/api/aggregatorTaskGasPriceBump", body); err != nil {
+ t.logger.Error("[Telemetry] Error in LogOperatorResponse", "error", err)
+	}
+}
+
 func (t *Telemetry) TaskSentToEthereum(batchMerkleRoot [32]byte, txHash string) {
 	body := TaskSentToEthereumMessage{
  MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),

config-files/config-aggregator.yaml

@@ -38,6 +38,9 @@ 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)
 
 ## Operator Configurations
 # operator:

core/chainio/avs_writer.go

@@ -15,7 +15,9 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
+	retry "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/config"
+	"github.com/yetanotherco/aligned_layer/core/utils"
 )
 
 type AvsWriter struct {
@@ -70,7 +72,10 @@ func NewAvsWriterFromConfig(baseConfig *config.BaseConfig, ecdsaConfig *config.E
 	}, nil
 }
 
-func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*common.Hash, error) {
+// 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.
+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
 	tx, err := w.RespondToTaskV2Retryable(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
@@ -83,17 +88,61 @@ func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMe
  return nil, err
 	}
 
-	// Send the transaction
+	// Set the nonce, as we might have to replace the transaction with a higher gas price
+	txNonce := big.NewInt(int64(tx.Nonce()))
+	txOpts.Nonce = txNonce
+	txOpts.GasPrice = tx.GasPrice()
 	txOpts.NoSend = false
-	txOpts.GasLimit = tx.Gas() * 110 / 100 // Add 10% to the gas limit
-	tx, err = w.RespondToTaskV2Retryable(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
-	if err != nil {
- return nil, err
+	i := 0
+
+	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
+ } else {
+ // bump the last tx gas price a little by `gasBumpIncrementalPercentage` to replace it.
+ txOpts.GasPrice = utils.CalculateGasPriceBumpBasedOnRetry(txOpts.GasPrice, gasBumpIncrementalPercentage, 0, 0)
+ }
+
+ if i > 0 {
+ onGasPriceBumped(txOpts.GasPrice)
+ }
+
+ err = w.checkRespondToTaskFeeLimit(tx, txOpts, batchIdentifierHash, senderAddress)
+ if err != nil {
+ return nil, retry.PermanentError{Inner: err}
+ }
+
+ w.logger.Infof("Sending RespondToTask transaction with a gas price of %v", txOpts.GasPrice)
+
+ tx, err = w.RespondToTaskV2Retryable(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+ if err != nil {
+ return nil, err
+ }
+
+ receipt, err := utils.WaitForTransactionReceiptRetryable(w.Client, w.ClientFallback, tx.Hash(), timeToWaitBeforeBump)
+ if receipt != nil {
+ return receipt, nil
+ }
+
+ // if we are here, it means we have reached the receipt waiting timeout
+ // 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...")
+ if err != nil {
+ return nil, err
+ }
+ return nil, fmt.Errorf("transaction failed")
 	}
 
-	txHash := tx.Hash()
-
-	return &txHash, nil
+	return retry.RetryWithData(respondToTaskV2Func, retry.MinDelay, retry.RetryFactor, 0, retry.MaxInterval, 0)
 }
 
 func (w *AvsWriter) checkRespondToTaskFeeLimit(tx *types.Transaction, txOpts bind.TransactOpts, batchIdentifierHash [32]byte, senderAddress [20]byte) error {

core/config/aggregator.go

@@ -25,6 +25,9 @@ type AggregatorConfig struct {
  GarbageCollectorTasksAge uint64
  GarbageCollectorTasksInterval uint64
  BlsServiceTaskTimeout time.Duration
+ GasBaseBumpPercentage uint
+ GasBumpIncrementalPercentage uint
+ TimeToWaitBeforeBump time.Duration
 	}
 }
 
@@ -40,6 +43,9 @@ 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"`
+ GasBaseBumpPercentage uint `yaml:"gas_base_bump_percentage"`
+ GasBumpIncrementalPercentage uint `yaml:"gas_bump_incremental_percentage"`
+ TimeToWaitBeforeBump time.Duration `yaml:"time_to_wait_before_bump"`
 	} `yaml:"aggregator"`
 }
 
@@ -85,6 +91,9 @@ func NewAggregatorConfig(configFilePath string) *AggregatorConfig {
  GarbageCollectorTasksAge uint64
  GarbageCollectorTasksInterval uint64
  BlsServiceTaskTimeout time.Duration
+ GasBaseBumpPercentage uint
+ GasBumpIncrementalPercentage uint
+ TimeToWaitBeforeBump time.Duration
  }(aggregatorConfigFromYaml.Aggregator),
 	}
 }

core/retry_test.go

@@ -152,7 +152,8 @@ func TestWaitForTransactionReceipt(t *testing.T) {
  t.Errorf("Error setting up Anvil: %s\n", err)
 	}
 
-	_, err = utils.WaitForTransactionReceipt(*client, context.Background(), hash)
+	// Assert Call succeeds when Anvil running
+	_, err = utils.WaitForTransactionReceiptRetryable(*client, *client, hash, time.Second*45)
 	assert.NotNil(t, err, "Error Waiting for Transaction with Anvil Running: %s\n", err)
 	if !strings.Contains(err.Error(), "not found") {
  t.Errorf("WaitForTransactionReceipt Emitted incorrect error: %s\n", err)
@@ -164,7 +165,7 @@ func TestWaitForTransactionReceipt(t *testing.T) {
  return
 	}
 
-	_, err = utils.WaitForTransactionReceipt(*client, context.Background(), hash)
+	_, err = utils.WaitForTransactionReceiptRetryable(*client, *client, hash, time.Second*45)
 	assert.NotNil(t, err)
 	if _, ok := err.(retry.PermanentError); ok {
  t.Errorf("WaitForTransactionReceipt Emitted non Transient error: %s\n", err)
@@ -180,7 +181,7 @@ func TestWaitForTransactionReceipt(t *testing.T) {
  t.Errorf("Error setting up Anvil: %s\n", err)
 	}
 
-	_, err = utils.WaitForTransactionReceipt(*client, context.Background(), hash)
+	_, err = utils.WaitForTransactionReceiptRetryable(*client, *client, hash, time.Second*45)
 	assert.NotNil(t, err)
 	if !strings.Contains(err.Error(), "not found") {
  t.Errorf("WaitForTransactionReceipt Emitted incorrect error: %s\n", err)

core/utils/eth_client_utils.go

@@ -2,6 +2,8 @@ package utils
 
 import (
 	"context"
+	"math/big"
+	"time"
 
 	"github.com/Layr-Labs/eigensdk-go/chainio/clients/eth"
 	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
@@ -10,11 +12,25 @@ import (
 	retry "github.com/yetanotherco/aligned_layer/core"
 )
 
-func WaitForTransactionReceipt(client eth.InstrumentedClient, ctx context.Context, txHash gethcommon.Hash) (*types.Receipt, error) {
+// WaitForTransactionReceiptRetryable repeatedly attempts to fetch the transaction receipt for a given transaction hash.
+// If the receipt is not found, the function will retry with exponential backoff until the specified `waitTimeout` duration is reached.
+// If the receipt is still unavailable after `waitTimeout`, it will return an error.
+//
+// Note: The `time.Second * 2` is set as the max interval in the retry mechanism because we can't reliably measure the specific time the tx will be included in a block.
+// Setting a higher value will imply doing less retries across the waitTimeout and so we might lose the receipt
+func WaitForTransactionReceiptRetryable(client eth.InstrumentedClient, fallbackClient eth.InstrumentedClient, txHash gethcommon.Hash, waitTimeout time.Duration) (*types.Receipt, error) {
 	receipt_func := func() (*types.Receipt, error) {
- return client.TransactionReceipt(ctx, txHash)
+ receipt, err := client.TransactionReceipt(context.Background(), txHash)
+ if err != nil {
+ receipt, err = client.TransactionReceipt(context.Background(), txHash)
+ if err != nil {
+ return nil, err
+ }
+ return receipt, nil
+ }
+ return receipt, nil
 	}
-	return retry.RetryWithData(receipt_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
+	return retry.RetryWithData(receipt_func, retry.MinDelay, retry.RetryFactor, 0, time.Second*2, waitTimeout)
 }
 
 func BytesToQuorumNumbers(quorumNumbersBytes []byte) eigentypes.QuorumNums {
@@ -32,3 +48,44 @@ func BytesToQuorumThresholdPercentages(quorumThresholdPercentagesBytes []byte) e
 	}
 	return quorumThresholdPercentages
 }
+
+// Simple algorithm to calculate the gasPrice bump based on:
+// the currentGasPrice, a base bump percentage, a retry percentage, and the retry count.
+// Formula: currentGasPrice + (currentGasPrice * (baseBumpPercentage + retryCount * incrementalRetryPercentage) / 100)
+func CalculateGasPriceBumpBasedOnRetry(currentGasPrice *big.Int, baseBumpPercentage uint, retryAttemptPercentage uint, retryCount int) *big.Int {
+	// Incremental percentage increase for each retry attempt (i*retryAttemptPercentage)
+	incrementalRetryPercentage := new(big.Int).Mul(big.NewInt(int64(retryAttemptPercentage)), big.NewInt(int64(retryCount)))
+
+	// Total bump percentage: base bump + incremental retry percentage
+	totalBumpPercentage := new(big.Int).Add(big.NewInt(int64(baseBumpPercentage)), incrementalRetryPercentage)
+
+	// Calculate the bump amount: currentGasPrice * totalBumpPercentage / 100
+	bumpAmount := new(big.Int).Mul(currentGasPrice, totalBumpPercentage)
+	bumpAmount = new(big.Int).Div(bumpAmount, big.NewInt(100))
+
+	// Final bumped gas price: currentGasPrice + bumpAmount
+	bumpedGasPrice := new(big.Int).Add(currentGasPrice, bumpAmount)
+
+	return bumpedGasPrice
+}
+
+/*
+GetGasPriceRetryable
+Get the gas price from the client with retry logic.
+- All errors are considered Transient Errors
+- Retry times: 1 sec, 2 sec, 4 sec
+*/
+func GetGasPriceRetryable(client eth.InstrumentedClient, fallbackClient eth.InstrumentedClient) (*big.Int, error) {
+	respondToTaskV2_func := func() (*big.Int, error) {
+ gasPrice, err := client.SuggestGasPrice(context.Background())
+ if err != nil {
+ gasPrice, err = fallbackClient.SuggestGasPrice(context.Background())
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return gasPrice, nil
+	}
+	return retry.RetryWithData(respondToTaskV2_func, retry.MinDelay, retry.RetryFactor, retry.NumRetries, retry.MaxInterval, retry.MaxElapsedTime)
+}

core/utils/eth_client_utils_test.go

@@ -0,0 +1,38 @@
+package utils_test
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/yetanotherco/aligned_layer/core/utils"
+)
+
+func TestCalculateGasPriceBumpBasedOnRetry(t *testing.T) {
+	baseBumpPercentage := uint(20)
+	incrementalRetryPercentage := uint(5)
+
+	gasPrices := [5]*big.Int{
+ big.NewInt(3000000000),
+ big.NewInt(3000000000),
+ big.NewInt(4000000000),
+ big.NewInt(4000000000),
+ big.NewInt(5000000000)}
+
+	expectedBumpedGasPrices := [5]*big.Int{
+ big.NewInt(3600000000),
+ big.NewInt(3750000000),
+ big.NewInt(5200000000),
+ big.NewInt(5400000000),
+ big.NewInt(7000000000)}
+
+	for i := 0; i < len(gasPrices); i++ {
+ currentGasPrice := gasPrices[i]
+ bumpedGasPrice := utils.CalculateGasPriceBumpBasedOnRetry(currentGasPrice, baseBumpPercentage, incrementalRetryPercentage, i)
+ expectedGasPrice := expectedBumpedGasPrices[i]
+
+ if bumpedGasPrice.Cmp(expectedGasPrice) != 0 {
+ t.Errorf("Bumped gas price does not match expected gas price, expected value %v, got: %v", expectedGasPrice, bumpedGasPrice)
+ }
+	}
+
+}

go.mod

@@ -12,6 +12,7 @@ require (
 
 require (
 	github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6
+	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/consensys/gnark v0.10.0
 	github.com/consensys/gnark-crypto v0.12.2-0.20240215234832-d72fcb379d3e
 	github.com/fxamacker/cbor/v2 v2.7.0