tx retry with higher gas price (#96)

* tx retry with higher gas price

Author: 0g-peterzhb

cmd/upload.go

@@ -56,6 +56,9 @@ type uploadArgument struct {
 	routines         int
 
 	fragmentSize int64
+	maxGasPrice  uint
+	nRetries     int
+	step         int64
 
 	timeout time.Duration
 }
@@ -79,6 +82,9 @@ func bindUploadFlags(cmd *cobra.Command, args *uploadArgument) {
 	cmd.Flags().Int64Var(&args.fragmentSize, "fragment-size", 1024*1024*1024*4, "the size of fragment to split into when file is too large")
 
 	cmd.Flags().IntVar(&args.routines, "routines", runtime.GOMAXPROCS(0), "number of go routines for uploading simutanously")
+	cmd.Flags().UintVar(&args.maxGasPrice, "max-gas-price", 0, "max gas price to send transaction")
+	cmd.Flags().IntVar(&args.nRetries, "n-retries", 0, "number of retries for uploading when it's not gas price issue")
+	cmd.Flags().Int64Var(&args.step, "step", 15, "step of gas price increasing, step / 10 (for 15, the new gas price is 1.5 * last gas price)")
 
 	cmd.Flags().DurationVar(&args.timeout, "timeout", 0, "cli task timeout, 0 for no timeout")
 }
@@ -124,6 +130,11 @@ func upload(*cobra.Command, []string) {
 	if uploadArgs.finalityRequired {
 		finalityRequired = transfer.FileFinalized
 	}
+
+	var maxGasPrice *big.Int
+	if uploadArgs.maxGasPrice > 0 {
+		maxGasPrice = big.NewInt(int64(uploadArgs.maxGasPrice))
+	}
 	opt := transfer.UploadOption{
 		Tags:             hexutil.MustDecode(uploadArgs.tags),
 		FinalityRequired: finalityRequired,
@@ -132,6 +143,9 @@ func upload(*cobra.Command, []string) {
 		SkipTx:           uploadArgs.skipTx,
 		Fee:              fee,
 		Nonce:            nonce,
+		MaxGasPrice:      maxGasPrice,
+		NRetries:         uploadArgs.nRetries,
+		Step:             uploadArgs.step,
 	}
 
 	file, err := core.Open(uploadArgs.file)

common/blockchain/rpc.go

@@ -20,6 +20,7 @@ import (
 var Web3LogEnabled bool
 
 type RetryOption struct {
+	NRetries int
 	Interval time.Duration
 	logger   *logrus.Logger
 }
@@ -70,12 +71,18 @@ func WaitForReceipt(ctx context.Context, client *web3go.Client, txHash common.Ha
 		opt.Interval = time.Second * 3
 	}
 
+	if opt.Interval == 0 {
+		opt.Interval = time.Second * 3
+	}
+
 	reminder := util.NewReminder(opt.logger, time.Minute)
 	for receipt == nil {
 		if receipt, err = client.WithContext(ctx).Eth.TransactionReceipt(txHash); err != nil {
 			return nil, err
 		}
 
+		logrus.WithField("txHash", txHash).WithField("receipt", receipt).Info("Transaction receipt")
+
 		// remind
 		if receipt == nil {
 			reminder.RemindWith("Transaction not executed yet", "hash", txHash)

contract/contract.go

@@ -10,9 +10,10 @@ import (
 	"github.com/0glabs/0g-storage-client/common/blockchain"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/openweb3/web3go"
+	"github.com/openweb3/web3go/types"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
@@ -26,11 +27,13 @@ type TxRetryOption struct {
 	Timeout          time.Duration
 	MaxNonGasRetries int
 	MaxGasPrice      *big.Int
+	Step             int64
 }
 
 var SpecifiedBlockError = "Specified block header does not exist"
-var DefaultTimeout = 30 * time.Second
-var DefaultMaxNonGasRetries = 10
+var DefaultTimeout = 15 * time.Minute
+var DefaultMaxNonGasRetries = 20
+var DefaultStep = int64(15)
 
 func IsRetriableSubmitLogEntryError(msg string) bool {
 	return strings.Contains(msg, SpecifiedBlockError) || strings.Contains(msg, "mempool") || strings.Contains(msg, "timeout")
@@ -52,6 +55,22 @@ func NewFlowContract(flowAddress common.Address, clientWithSigner *web3go.Client
 	return &FlowContract{contract, flow, clientWithSigner}, nil
 }
 
+func (f *FlowContract) GetNonce(ctx context.Context) (*big.Int, error) {
+	sm, err := f.clientWithSigner.GetSignerManager()
+	if err != nil {
+		return nil, err
+	}
+
+	addr := sm.List()[0].Address()
+
+	nonce, err := f.clientWithSigner.Eth.TransactionCount(addr, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return nonce, nil
+}
+
 func (f *FlowContract) GetGasPrice() (*big.Int, error) {
 	gasPrice, err := f.clientWithSigner.Eth.GasPrice()
 	if err != nil {
@@ -110,68 +129,161 @@ func TransactWithGasAdjustment(
 	opts *bind.TransactOpts,
 	retryOpts *TxRetryOption,
 	params ...interface{},
-) (*types.Transaction, error) {
+) (*types.Receipt, error) {
 	// Set timeout and max non-gas retries from retryOpts if provided.
 	if retryOpts == nil {
 		retryOpts = &TxRetryOption{
-			Timeout:          DefaultTimeout,
 			MaxNonGasRetries: DefaultMaxNonGasRetries,
 		}
 	}
 
+	if retryOpts.MaxNonGasRetries == 0 {
+		retryOpts.MaxNonGasRetries = DefaultMaxNonGasRetries
+	}
+
+	if retryOpts.Step == 0 {
+		retryOpts.Step = DefaultStep
+	}
+
+	if t, ok := opts.Context.Deadline(); ok {
+		retryOpts.Timeout = time.Until(t)
+	}
+
+	logrus.WithField("timeout", retryOpts.Timeout).WithField("maxNonGasRetries", retryOpts.MaxNonGasRetries).Debug("Set retry options")
+
+	if opts.Nonce == nil {
+		// Get the current nonce if not set.
+		nonce, err := contract.GetNonce(opts.Context)
+		if err != nil {
+			return nil, err
+		}
+		// add one to the nonce
+		opts.Nonce = nonce
+	}
+
+	logrus.WithField("nonce", opts.Nonce).Info("Set nonce")
+
 	if opts.GasPrice == nil {
 		// Get the current gas price if not set.
 		gasPrice, err := contract.GetGasPrice()
 		if err != nil {
-			return nil, fmt.Errorf("failed to get gas price: %w", err)
+			return nil, errors.WithMessage(err, "failed to get gas price")
 		}
 		opts.GasPrice = gasPrice
 		logrus.WithField("gasPrice", opts.GasPrice).Debug("Receive current gas price from chain node")
 	}
 
 	logrus.WithField("gasPrice", opts.GasPrice).Info("Set gas price")
 
-	nRetries := 0
-	for {
-		// Create a fresh context per iteration.
-		ctx, cancel := context.WithTimeout(context.Background(), retryOpts.Timeout)
-		opts.Context = ctx
-		tx, err := contract.FlowTransactor.contract.Transact(opts, method, params...)
-		cancel() // cancel this iteration's context
-		if err == nil {
-			return tx, nil
-		}
+	receiptCh := make(chan *types.Receipt, 1)
+	errCh := make(chan error, 1)
+	failCh := make(chan error, 1)
 
-		errStr := strings.ToLower(err.Error())
+	var ctx context.Context
+	var cancel context.CancelFunc
+	if retryOpts.Timeout > 0 {
+		ctx, cancel = context.WithTimeout(context.Background(), retryOpts.Timeout)
+	} else {
+		ctx, cancel = context.WithCancel(context.Background())
+	}
 
-		if !IsRetriableSubmitLogEntryError(errStr) {
-			return nil, fmt.Errorf("failed to send transaction: %w", err)
+	// calculate number of gas retry by dividing max gas price by current gas price and the ration
+	nGasRetry := 0
+	if retryOpts.MaxGasPrice != nil {
+		gasPrice := opts.GasPrice
+		for gasPrice.Cmp(retryOpts.MaxGasPrice) <= 0 {
+			gasPrice = new(big.Int).Mul(gasPrice, big.NewInt(retryOpts.Step))
+			gasPrice.Div(gasPrice, big.NewInt(10))
+			nGasRetry++
 		}
+	}
 
-		if strings.Contains(errStr, "mempool") || strings.Contains(errStr, "timeout") {
-			if retryOpts.MaxGasPrice == nil {
-				return nil, fmt.Errorf("mempool full and no max gas price is set, failed to send transaction: %w", err)
-			} else {
-				newGasPrice := new(big.Int).Mul(opts.GasPrice, big.NewInt(11))
-				newGasPrice.Div(newGasPrice, big.NewInt(10))
-				if newGasPrice.Cmp(retryOpts.MaxGasPrice) > 0 {
-					opts.GasPrice = new(big.Int).Set(retryOpts.MaxGasPrice)
-				} else {
-					opts.GasPrice = newGasPrice
+	go func() {
+		nRetries := 0
+		for {
+			select {
+			case <-ctx.Done():
+				// main or another goroutine canceled the context
+				logrus.Info("Context canceled; stopping outer loop")
+				return
+			default:
+			}
+			tx, err := contract.FlowTransactor.contract.Transact(opts, method, params...)
+
+			var receipt *types.Receipt
+			if err == nil {
+				// Wait for successful execution
+				go func() {
+					receipt, err = contract.WaitForReceipt(ctx, tx.Hash(), true, blockchain.RetryOption{NRetries: retryOpts.MaxNonGasRetries})
+					if err == nil {
+						receiptCh <- receipt
+						return
+					}
+					errCh <- err
+				}()
+				// even if the receipt is received, this loop will continue until the context is canceled
+				time.Sleep(30 * time.Second)
+				err = fmt.Errorf("timeout")
+			}
+
+			logrus.WithError(err).Error("Failed to send transaction")
+
+			errStr := strings.ToLower(err.Error())
+
+			if !IsRetriableSubmitLogEntryError(errStr) {
+				if strings.Contains(errStr, "invalid nonce") {
+					return
 				}
-				logrus.WithError(err).Infof("Increasing gas price to %v due to mempool/timeout error", opts.GasPrice)
+				failCh <- errors.WithMessage(err, "failed to send transaction")
+				return
 			}
-		} else {
-			nRetries++
-			if nRetries >= retryOpts.MaxNonGasRetries {
-				return nil, fmt.Errorf("failed to send transaction after %d retries: %w", nRetries, err)
+
+			// If the error is due to mempool full or timeout, retry with a higher gas price
+			if strings.Contains(errStr, "mempool") || strings.Contains(errStr, "timeout") {
+				if retryOpts.MaxGasPrice == nil {
+					failCh <- errors.WithMessage(err, "mempool full and no max gas price is set, failed to send transaction")
+					return
+				} else if opts.GasPrice.Cmp(retryOpts.MaxGasPrice) >= 0 {
+					return
+				} else {
+					newGasPrice := new(big.Int).Mul(opts.GasPrice, big.NewInt(retryOpts.Step))
+					newGasPrice.Div(newGasPrice, big.NewInt(10))
+					if newGasPrice.Cmp(retryOpts.MaxGasPrice) > 0 {
+						opts.GasPrice = new(big.Int).Set(retryOpts.MaxGasPrice)
+					} else {
+						opts.GasPrice = newGasPrice
+					}
+					logrus.WithError(err).Infof("Increasing gas price to %v due to mempool/timeout error", opts.GasPrice)
+				}
+			} else {
+				nRetries++
+				if nRetries >= retryOpts.MaxNonGasRetries {
+					failCh <- errors.WithMessage(err, "failed to send transaction")
+					return
+				}
+				logrus.WithError(err).Infof("Retrying with same gas price %v, attempt %d", opts.GasPrice, nRetries)
 			}
-			logrus.WithError(err).Infof("Retrying with same gas price %v, attempt %d", opts.GasPrice, nRetries)
 		}
+	}()
 
-		time.Sleep(10 * time.Second)
+	nErr := 0
+	for {
+		select {
+		case receipt := <-receiptCh:
+			cancel()
+			return receipt, nil
+		case err := <-errCh:
+			nErr++
+			if nErr >= nGasRetry {
+				failCh <- errors.WithMessage(err, "All gas price retries failed")
+				cancel()
+				return nil, err
+			}
+		case err := <-failCh:
+			cancel()
+			return nil, err
+		}
 	}
-
 }
 
 func (submission Submission) Fee(pricePerSector *big.Int) *big.Int {

tests/go_tests/segment_upload_test/main.go

@@ -98,9 +98,12 @@ func runTest() error {
 	if err != nil {
 		return errors.WithMessage(err, "failed to initialize uploader")
 	}
-	_, _, err = uploader.SubmitLogEntry(ctx, []core.IterableData{data}, make([][]byte, 1), nil, nil)
+	_, _, err = uploader.SubmitLogEntry(ctx, []core.IterableData{data}, make([][]byte, 1), transfer.SubmitLogEntryOption{
+		NRetries: 5,
+		Step:     15,
+	})
 	if err != nil {
-		return errors.WithMessage(err, "failed to sub log entry")
+		return errors.WithMessage(err, "failed to submit log entry")
 	}
 	// wait for log entry
 	var info *node.FileInfo