feat: aggregator bump fee if transaction was not included

aggregator/internal/pkg/aggregator.go

@@ -304,7 +304,8 @@ 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)
+	onRetry := func() { agg.metrics.IncBumpedGasPriceForAggregatedResponse() }
+	receipt, err := agg.avsWriter.SendAggregatedResponse(batchIdentifierHash, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature, onRetry)
 	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)
@@ -315,13 +316,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

core/chainio/avs_writer.go

@@ -15,7 +15,15 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
+	connection "github.com/yetanotherco/aligned_layer/core"
 	"github.com/yetanotherco/aligned_layer/core/config"
+	"github.com/yetanotherco/aligned_layer/core/utils"
+)
+
+const (
+	GasBumpPercentage int = 20
+	// wait as much as 3 blocks time for the receipt
+	SendAggregateResponseReceiptTimeout time.Duration = time.Second * 36
 )
 
 type AvsWriter struct {
@@ -70,10 +78,15 @@ 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.
+// Note: If the rpc endpoints fail, the retry will stop, as it will infinitely try
+func (w *AvsWriter) SendAggregatedResponse(batchIdentifierHash [32]byte, batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature, onRetry func()) (*types.Receipt, error) {
 	txOpts := *w.Signer.GetTxOpts()
 	txOpts.NoSend = true // simulate the transaction
 	tx, err := w.AvsContractBindings.ServiceManager.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+
 	if err != nil {
 		// Retry with fallback
 		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
@@ -87,21 +100,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 fee
+	txNonce := new(big.Int).SetUint64(tx.Nonce())
 	txOpts.NoSend = false
-	txOpts.GasLimit = tx.Gas() * 110 / 100 // Add 10% to the gas limit
-	tx, err = w.AvsContractBindings.ServiceManager.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
-	if err != nil {
-		// Retry with fallback
-		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	txOpts.Nonce = txNonce
+
+	i := 0
+	sendTransaction := func() (*types.Receipt, error) {
+		if i > 0 {
+			onRetry()
+		}
+		i++
+
+		// get gas price
+		gasPrice, err := w.Client.SuggestGasPrice(context.Background())
+		if err != nil {
+			// Retry with fallback
+			gasPrice, err = w.ClientFallback.SuggestGasPrice(context.Background())
+			if err != nil {
+				return nil, fmt.Errorf("transaction simulation failed: %v", err)
+			}
+		}
+		txOpts.GasPrice = utils.CalculateGasPriceBumpBasedOnRetry(gasPrice, GasBumpPercentage, i)
+
+		w.logger.Infof("Sending ResponseToTask transaction with a gas price of %v", txOpts.GasPrice)
+		err = w.checkRespondToTaskFeeLimit(tx, txOpts, batchIdentifierHash, senderAddress)
+
+		if err != nil {
+			return nil, connection.PermanentError{Inner: err}
+		}
+
+		tx, err = w.AvsContractBindings.ServiceManager.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+		if err != nil {
+			// Retry with fallback
+			tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+			if err != nil {
+				return nil, connection.PermanentError{Inner: err}
+			}
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), SendAggregateResponseReceiptTimeout)
+		defer cancel()
+		receipt, err := utils.WaitForTransactionReceipt(w.Client, ctx, tx.Hash())
+
+		if receipt != nil {
+			return receipt, nil
+		}
+		// if we are here, this means we have reached the timeout (after three blocks it hasn't been included)
+		// so we try again by bumping the fee to make sure its included
 		if err != nil {
 			return nil, err
 		}
+		return nil, fmt.Errorf("transaction failed")
 	}
 
-	txHash := tx.Hash()
-
-	return &txHash, nil
+	return connection.RetryWithData(sendTransaction, 1000, 2, 0)
 }
 
 func (w *AvsWriter) checkRespondToTaskFeeLimit(tx *types.Transaction, txOpts bind.TransactOpts, batchIdentifierHash [32]byte, senderAddress [20]byte) error {

core/connection.go

@@ -0,0 +1,77 @@
+package connection
+
+import (
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+)
+
+type PermanentError struct {
+	Inner error
+}
+
+func (e PermanentError) Error() string { return e.Inner.Error() }
+func (e PermanentError) Unwrap() error {
+	return e.Inner
+}
+func (e PermanentError) Is(err error) bool {
+	_, ok := err.(PermanentError)
+	return ok
+}
+
+// Same as Retry only that the functionToRetry can return a value upon correct execution
+func RetryWithData[T any](functionToRetry func() (*T, error), minDelay uint64, factor float64, maxTries uint64) (*T, error) {
+	f := func() (*T, error) {
+		val, err := functionToRetry()
+		if perm, ok := err.(PermanentError); err != nil && ok {
+			return nil, backoff.Permanent(perm.Inner)
+		}
+		return val, err
+	}
+
+	randomOption := backoff.WithRandomizationFactor(0)
+
+	initialRetryOption := backoff.WithInitialInterval(time.Millisecond * time.Duration(minDelay))
+	multiplierOption := backoff.WithMultiplier(factor)
+	expBackoff := backoff.NewExponentialBackOff(randomOption, multiplierOption, initialRetryOption)
+	var maxRetriesBackoff backoff.BackOff
+
+	if maxTries > 0 {
+		maxRetriesBackoff = backoff.WithMaxRetries(expBackoff, maxTries)
+	} else {
+		maxRetriesBackoff = expBackoff
+	}
+
+	return backoff.RetryWithData(f, maxRetriesBackoff)
+}
+
+// Retries a given function in an exponential backoff manner.
+// It will retry calling the function while it returns an error, until the max retries.
+// If maxTries == 0 then the retry function will run indefinitely until success
+// from the configuration are reached, or until a `PermanentError` is returned.
+// The function to be retried should return `PermanentError` when the condition for stop retrying
+// is met.
+func Retry(functionToRetry func() error, minDelay uint64, factor float64, maxTries uint64) error {
+	f := func() error {
+		err := functionToRetry()
+		if perm, ok := err.(PermanentError); err != nil && ok {
+			return backoff.Permanent(perm.Inner)
+		}
+		return err
+	}
+
+	randomOption := backoff.WithRandomizationFactor(0)
+
+	initialRetryOption := backoff.WithInitialInterval(time.Millisecond * time.Duration(minDelay))
+	multiplierOption := backoff.WithMultiplier(factor)
+	expBackoff := backoff.NewExponentialBackOff(randomOption, multiplierOption, initialRetryOption)
+	var maxRetriesBackoff backoff.BackOff
+
+	if maxTries > 0 {
+		maxRetriesBackoff = backoff.WithMaxRetries(expBackoff, maxTries)
+	} else {
+		maxRetriesBackoff = expBackoff
+	}
+
+	return backoff.Retry(f, maxRetriesBackoff)
+}

core/connection_test.go

@@ -0,0 +1,81 @@
+package connection_test
+
+import (
+	"fmt"
+	"testing"
+
+	connection "github.com/yetanotherco/aligned_layer/core"
+)
+
+func DummyFunction(x uint64) (uint64, error) {
+	fmt.Println("Doing some work...")
+	if x == 42 {
+		return 0, connection.PermanentError{Inner: fmt.Errorf("Permanent error!")}
+	} else if x < 42 {
+		return 0, fmt.Errorf("Transient error!")
+	}
+	return x, nil
+}
+
+// here we run the dummy function and with the retry and check:
+// - The number of retries checks based on the `n`
+// - The returned valued matches based on the `n`
+// - The returned err matches based on the `n`
+func TestRetryWithData(t *testing.T) {
+	retries := -1
+	testFun := func(n uint64) func() (*uint64, error) {
+		return func() (*uint64, error) {
+			retries++
+			x, err := DummyFunction(n)
+			return &x, err
+		}
+	}
+	data, err := connection.RetryWithData(testFun(uint64(41)), 1000, 2, 3)
+	if !(retries == 3 && *data == 0 && err != nil) {
+		t.Error("Incorrect execution when n == 41")
+	}
+	//restart
+	retries = -1
+	data, err = connection.RetryWithData(testFun(42), 1000, 2, 3)
+	if !(retries == 0 && data == nil) {
+		if _, ok := err.(*connection.PermanentError); ok {
+			t.Error("Incorrect execution when n == 42")
+		}
+	}
+	//restart
+	retries = -1
+	data, err = connection.RetryWithData(testFun(43), 1000, 2, 3)
+	if !(retries == 0 && *data == 43 && err == nil) {
+		t.Error("Incorrect execution when n == 43")
+	}
+}
+
+// same as above but without checking returned data
+func TestRetry(t *testing.T) {
+	retries := -1
+	testFun := func(n uint64) func() error {
+		return func() error {
+			retries++
+			_, err := DummyFunction(n)
+			return err
+		}
+	}
+	err := connection.Retry(testFun(uint64(41)), 1000, 2, 3)
+	if !(retries == 3 && err != nil) {
+		t.Error("Incorrect execution when n == 41")
+	}
+	//restart
+	retries = -1
+	err = connection.Retry(testFun(42), 1000, 2, 3)
+	if !(retries == 0) {
+		if _, ok := err.(*connection.PermanentError); ok {
+			t.Error("Incorrect execution when n == 42")
+		}
+	}
+	//restart
+	retries = -1
+	err = connection.Retry(testFun(43), 1000, 2, 3)
+	if !(retries == 0 && err == nil) {
+		t.Error("Incorrect execution when n == 43")
+	}
+}

core/utils/eth_client_utils.go

@@ -2,6 +2,7 @@ package utils
 
 import (
 	"context"
+	"math/big"
 	"time"
 
 	"fmt"
@@ -17,12 +18,16 @@ const sleepTime = 5 * time.Second
 
 func WaitForTransactionReceipt(client eth.InstrumentedClient, ctx context.Context, txHash gethcommon.Hash) (*types.Receipt, error) {
 	for i := 0; i < maxRetries; i++ {
-		receipt, err := client.TransactionReceipt(ctx, txHash)
-		if err != nil {
-			time.Sleep(sleepTime)
-		} else {
+		receipt, _ := client.TransactionReceipt(ctx, txHash)
+		if receipt != nil {
 			return receipt, nil
 		}
+		// if context has timed out, return
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		} else {
+			time.Sleep(sleepTime)
+		}
 	}
 	return nil, fmt.Errorf("transaction receipt not found for txHash: %s", txHash.String())
 }
@@ -42,3 +47,15 @@ func BytesToQuorumThresholdPercentages(quorumThresholdPercentagesBytes []byte) e
 	}
 	return quorumThresholdPercentages
 }
+
+// Very basic algorithm to calculate the gasPrice bump based on the currentGasPrice a constant percentage and the retry number.
+// It adds a the percentage to the current gas price and a 5% * i, where i is the iteration number. That is:
+func CalculateGasPriceBumpBasedOnRetry(currentGasPrice *big.Int, percentage int, i int) *big.Int {
+	retryPercentage := new(big.Int).Mul(big.NewInt(5), big.NewInt(int64(i)))
+	percentageBump := new(big.Int).Add(big.NewInt(int64(percentage)), retryPercentage)
+	bumpAmount := new(big.Int).Mul(currentGasPrice, percentageBump)
+	bumpAmount = new(big.Int).Div(bumpAmount, big.NewInt(100))
+	bumpedGasPrice := new(big.Int).Add(currentGasPrice, bumpAmount)
+
+	return 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

go.sum

@@ -54,8 +54,8 @@ github.com/btcsuite/btcd/btcec/v2 v2.3.4 h1:3EJjcN70HCu/mwqlUsGK8GcNVyLVxFDlWurT
 github.com/btcsuite/btcd/btcec/v2 v2.3.4/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
 github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 h1:q0rUy8C/TYNBQS1+CGKw68tLOFYSNEs0TFnxxnS9+4U=
 github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1/go.mod h1:7SFka0XMvUgj3hfZtydOrQY2mwhPclbT2snogU7SQQc=
-github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
-github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/cespare/cp v0.1.0 h1:SE+dxFebS7Iik5LK0tsi1k9ZCxEaFX4AjQmoyA+1dJk=
 github.com/cespare/cp v0.1.0/go.mod h1:SOGHArjBr4JWaSDEVpWpo/hNg6RoKrls6Oh40hiwW+s=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=