Merge pull request #7472 from mdehoog/michael/retry-underpriced-immediately

txmgr: don't wait the resubmission timeout for gas increases for underpriced errors

op-e2e/setup.go

@@ -71,6 +71,7 @@ func newTxMgrConfig(l1Addr string, privKey *ecdsa.PrivateKey) txmgr.CLIConfig {
 		PrivateKey:                hexPriv(privKey),
 		NumConfirmations:          1,
 		SafeAbortNonceTooLowCount: 3,
+		FeeLimitMultiplier:        5,
 		ResubmissionTimeout:       3 * time.Second,
 		ReceiptQueryInterval:      50 * time.Millisecond,
 		NetworkTimeout:            2 * time.Second,

op-service/txmgr/cli.go

@@ -26,6 +26,7 @@ const (
 	// TxMgr Flags (new + legacy + some shared flags)
 	NumConfirmationsFlagName          = "num-confirmations"
 	SafeAbortNonceTooLowCountFlagName = "safe-abort-nonce-too-low-count"
+	FeeLimitMultiplierFlagName        = "fee-limit-multiplier"
 	ResubmissionTimeoutFlagName       = "resubmission-timeout"
 	NetworkTimeoutFlagName            = "network-timeout"
 	TxSendTimeoutFlagName             = "txmgr.send-timeout"
@@ -51,6 +52,7 @@ var (
 type DefaultFlagValues struct {
 	NumConfirmations          uint64
 	SafeAbortNonceTooLowCount uint64
+	FeeLimitMultiplier        uint64
 	ResubmissionTimeout       time.Duration
 	NetworkTimeout            time.Duration
 	TxSendTimeout             time.Duration
@@ -62,6 +64,7 @@ var (
 	DefaultBatcherFlagValues = DefaultFlagValues{
 		NumConfirmations:          uint64(10),
 		SafeAbortNonceTooLowCount: uint64(3),
+		FeeLimitMultiplier:        uint64(5),
 		ResubmissionTimeout:       48 * time.Second,
 		NetworkTimeout:            10 * time.Second,
 		TxSendTimeout:             0 * time.Second,
@@ -71,6 +74,7 @@ var (
 	DefaultChallengerFlagValues = DefaultFlagValues{
 		NumConfirmations:          uint64(3),
 		SafeAbortNonceTooLowCount: uint64(3),
+		FeeLimitMultiplier:        uint64(5),
 		ResubmissionTimeout:       24 * time.Second,
 		NetworkTimeout:            10 * time.Second,
 		TxSendTimeout:             2 * time.Minute,
@@ -115,6 +119,12 @@ func CLIFlagsWithDefaults(envPrefix string, defaults DefaultFlagValues) []cli.Fl
 			Value:   defaults.SafeAbortNonceTooLowCount,
 			EnvVars: prefixEnvVars("SAFE_ABORT_NONCE_TOO_LOW_COUNT"),
 		},
+		&cli.Uint64Flag{
+			Name:    FeeLimitMultiplierFlagName,
+			Usage:   "The multiplier applied to fee suggestions to put a hard limit on fee increases",
+			Value:   defaults.FeeLimitMultiplier,
+			EnvVars: prefixEnvVars("TXMGR_FEE_LIMIT_MULTIPLIER"),
+		},
 		&cli.DurationFlag{
 			Name:    ResubmissionTimeoutFlagName,
 			Usage:   "Duration we will wait before resubmitting a transaction to L1",
@@ -158,6 +168,7 @@ type CLIConfig struct {
 	SignerCLIConfig           opsigner.CLIConfig
 	NumConfirmations          uint64
 	SafeAbortNonceTooLowCount uint64
+	FeeLimitMultiplier        uint64
 	ResubmissionTimeout       time.Duration
 	ReceiptQueryInterval      time.Duration
 	NetworkTimeout            time.Duration
@@ -170,6 +181,7 @@ func NewCLIConfig(l1RPCURL string, defaults DefaultFlagValues) CLIConfig {
 		L1RPCURL:                  l1RPCURL,
 		NumConfirmations:          defaults.NumConfirmations,
 		SafeAbortNonceTooLowCount: defaults.SafeAbortNonceTooLowCount,
+		FeeLimitMultiplier:        defaults.FeeLimitMultiplier,
 		ResubmissionTimeout:       defaults.ResubmissionTimeout,
 		NetworkTimeout:            defaults.NetworkTimeout,
 		TxSendTimeout:             defaults.TxSendTimeout,
@@ -189,6 +201,9 @@ func (m CLIConfig) Check() error {
 	if m.NetworkTimeout == 0 {
 		return errors.New("must provide NetworkTimeout")
 	}
+	if m.FeeLimitMultiplier == 0 {
+		return errors.New("must provide FeeLimitMultiplier")
+	}
 	if m.ResubmissionTimeout == 0 {
 		return errors.New("must provide ResubmissionTimeout")
 	}
@@ -218,6 +233,7 @@ func ReadCLIConfig(ctx *cli.Context) CLIConfig {
 		SignerCLIConfig:           opsigner.ReadCLIConfig(ctx),
 		NumConfirmations:          ctx.Uint64(NumConfirmationsFlagName),
 		SafeAbortNonceTooLowCount: ctx.Uint64(SafeAbortNonceTooLowCountFlagName),
+		FeeLimitMultiplier:        ctx.Uint64(FeeLimitMultiplierFlagName),
 		ResubmissionTimeout:       ctx.Duration(ResubmissionTimeoutFlagName),
 		ReceiptQueryInterval:      ctx.Duration(ReceiptQueryIntervalFlagName),
 		NetworkTimeout:            ctx.Duration(NetworkTimeoutFlagName),
@@ -261,6 +277,7 @@ func NewConfig(cfg CLIConfig, l log.Logger) (Config, error) {
 	return Config{
 		Backend:                   l1,
 		ResubmissionTimeout:       cfg.ResubmissionTimeout,
+		FeeLimitMultiplier:        cfg.FeeLimitMultiplier,
 		ChainID:                   chainID,
 		TxSendTimeout:             cfg.TxSendTimeout,
 		TxNotInMempoolTimeout:     cfg.TxNotInMempoolTimeout,
@@ -282,6 +299,9 @@ type Config struct {
 	// attempted.
 	ResubmissionTimeout time.Duration
 
+	// The multiplier applied to fee suggestions to put a hard limit on fee increases.
+	FeeLimitMultiplier uint64
+
 	// ChainID is the chain ID of the L1 chain.
 	ChainID *big.Int
 
@@ -326,6 +346,9 @@ func (m Config) Check() error {
 	if m.NetworkTimeout == 0 {
 		return errors.New("must provide NetworkTimeout")
 	}
+	if m.FeeLimitMultiplier == 0 {
+		return errors.New("must provide FeeLimitMultiplier")
+	}
 	if m.ResubmissionTimeout == 0 {
 		return errors.New("must provide ResubmissionTimeout")
 	}

op-service/txmgr/send_state.go

@@ -26,6 +26,9 @@ type SendState struct {
 	// Counts of the different types of errors
 	successFullPublishCount   uint64 // nil error => tx made it to the mempool
 	safeAbortNonceTooLowCount uint64 // nonce too low error
+
+	// Miscellaneous tracking
+	bumpCount int // number of times we have bumped the gas price
 }
 
 // NewSendStateWithNow creates a new send state with the provided clock.

op-service/txmgr/txmgr.go

@@ -24,9 +24,6 @@ import (
 const (
 	// Geth requires a minimum fee bump of 10% for tx resubmission
 	priceBump int64 = 10
-
-	// The multiplier applied to fee suggestions to put a hard limit on fee increases
-	feeLimitMultiplier = 5
 )
 
 // new = old * (100 + priceBump) / 100
@@ -297,19 +294,26 @@ func (m *SimpleTxManager) sendTx(ctx context.Context, tx *types.Transaction) (*t
 
 	sendState := NewSendState(m.cfg.SafeAbortNonceTooLowCount, m.cfg.TxNotInMempoolTimeout)
 	receiptChan := make(chan *types.Receipt, 1)
-	sendTxAsync := func(tx *types.Transaction) {
-		defer wg.Done()
-		m.publishAndWaitForTx(ctx, tx, sendState, receiptChan)
+	publishAndWait := func(tx *types.Transaction, bumpFees bool) *types.Transaction {
+		wg.Add(1)
+		tx, published := m.publishTx(ctx, tx, sendState, bumpFees)
+		if published {
+			go func() {
+				defer wg.Done()
+				m.waitForTx(ctx, tx, sendState, receiptChan)
+			}()
+		} else {
+			wg.Done()
+		}
+		return tx
 	}
 
 	// Immediately publish a transaction before starting the resumbission loop
-	wg.Add(1)
-	go sendTxAsync(tx)
+	tx = publishAndWait(tx, false)
 
 	ticker := time.NewTicker(m.cfg.ResubmissionTimeout)
 	defer ticker.Stop()
 
-	bumpCounter := 0
 	for {
 		select {
 		case <-ticker.C:
@@ -322,73 +326,95 @@ func (m *SimpleTxManager) sendTx(ctx context.Context, tx *types.Transaction) (*t
 				m.l.Warn("Aborting transaction submission")
 				return nil, errors.New("aborted transaction sending")
 			}
-			// Increase the gas price & submit the new transaction
-			newTx, err := m.increaseGasPrice(ctx, tx)
-			if err != nil || sendState.IsWaitingForConfirmation() {
-				// there is a chance the previous tx goes into "waiting for confirmation" state
-				// during the increaseGasPrice call. In some (but not all) cases increaseGasPrice
-				// will error out during gas estimation. In either case we should continue waiting
-				// rather than resubmit the tx.
-				continue
-			}
-			tx = newTx
-			wg.Add(1)
-			bumpCounter += 1
-			go sendTxAsync(tx)
+			tx = publishAndWait(tx, true)
 
 		case <-ctx.Done():
 			return nil, ctx.Err()
 
 		case receipt := <-receiptChan:
-			m.metr.RecordGasBumpCount(bumpCounter)
+			m.metr.RecordGasBumpCount(sendState.bumpCount)
 			m.metr.TxConfirmed(receipt)
 			return receipt, nil
 		}
 	}
 }
 
-// publishAndWaitForTx publishes the transaction to the transaction pool and then waits for it with [waitMined].
-// It should be called in a new go-routine. It will send the receipt to receiptChan in a non-blocking way if a receipt is found
-// for the transaction.
-func (m *SimpleTxManager) publishAndWaitForTx(ctx context.Context, tx *types.Transaction, sendState *SendState, receiptChan chan *types.Receipt) {
-	log := m.l.New("hash", tx.Hash(), "nonce", tx.Nonce(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
-	log.Info("Publishing transaction")
+// publishTx publishes the transaction to the transaction pool. If it receives any underpriced errors
+// it will bump the fees and retry.
+// Returns the latest fee bumped tx, and a boolean indicating whether the tx was sent or not
+func (m *SimpleTxManager) publishTx(ctx context.Context, tx *types.Transaction, sendState *SendState, bumpFeesImmediately bool) (*types.Transaction, bool) {
+	updateLogFields := func(tx *types.Transaction) log.Logger {
+		return m.l.New("hash", tx.Hash(), "nonce", tx.Nonce(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
+	}
+	l := updateLogFields(tx)
 
-	cCtx, cancel := context.WithTimeout(ctx, m.cfg.NetworkTimeout)
-	defer cancel()
-	t := time.Now()
-	err := m.backend.SendTransaction(cCtx, tx)
-	sendState.ProcessSendError(err)
+	l.Info("Publishing transaction")
+
+	for {
+		if bumpFeesImmediately {
+			newTx, err := m.increaseGasPrice(ctx, tx)
+			if err != nil {
+				l.Error("unable to increase gas", "err", err)
+				m.metr.TxPublished("bump_failed")
+				return tx, false
+			}
+			tx = newTx
+			sendState.bumpCount++
+			l = updateLogFields(tx)
+		}
+		bumpFeesImmediately = true // bump fees next loop
+
+		if sendState.IsWaitingForConfirmation() {
+			// there is a chance the previous tx goes into "waiting for confirmation" state
+			// during the increaseGasPrice call; continue waiting rather than resubmit the tx
+			return tx, false
+		}
+
+		cCtx, cancel := context.WithTimeout(ctx, m.cfg.NetworkTimeout)
+		err := m.backend.SendTransaction(cCtx, tx)
+		cancel()
+		sendState.ProcessSendError(err)
+
+		if err == nil {
+			m.metr.TxPublished("")
+			log.Info("Transaction successfully published")
+			return tx, true
+		}
 
-	// Properly log & exit if there is an error
-	if err != nil {
 		switch {
 		case errStringMatch(err, core.ErrNonceTooLow):
-			log.Warn("nonce too low", "err", err)
+			l.Warn("nonce too low", "err", err)
 			m.metr.TxPublished("nonce_to_low")
 		case errStringMatch(err, context.Canceled):
 			m.metr.RPCError()
-			log.Warn("transaction send cancelled", "err", err)
+			l.Warn("transaction send cancelled", "err", err)
 			m.metr.TxPublished("context_cancelled")
 		case errStringMatch(err, txpool.ErrAlreadyKnown):
-			log.Warn("resubmitted already known transaction", "err", err)
+			l.Warn("resubmitted already known transaction", "err", err)
 			m.metr.TxPublished("tx_already_known")
 		case errStringMatch(err, txpool.ErrReplaceUnderpriced):
-			log.Warn("transaction replacement is underpriced", "err", err)
+			l.Warn("transaction replacement is underpriced", "err", err)
 			m.metr.TxPublished("tx_replacement_underpriced")
+			continue // retry with fee bump
 		case errStringMatch(err, txpool.ErrUnderpriced):
-			log.Warn("transaction is underpriced", "err", err)
+			l.Warn("transaction is underpriced", "err", err)
 			m.metr.TxPublished("tx_underpriced")
+			continue // retry with fee bump
 		default:
 			m.metr.RPCError()
-			log.Error("unable to publish transaction", "err", err)
+			l.Error("unable to publish transaction", "err", err)
 			m.metr.TxPublished("unknown_error")
 		}
-		return
+
+		// on non-underpriced error return immediately; will retry on next resubmission timeout
+		return tx, false
 	}
-	m.metr.TxPublished("")
+}
 
-	log.Info("Transaction successfully published")
+// waitForTx calls waitMined, and then sends the receipt to receiptChan in a non-blocking way if a receipt is found
+// for the transaction. It should be called in a separate goroutine.
+func (m *SimpleTxManager) waitForTx(ctx context.Context, tx *types.Transaction, sendState *SendState, receiptChan chan *types.Receipt) {
+	t := time.Now()
 	// Poll for the transaction to be ready & then send the result to receiptChan
 	receipt, err := m.waitMined(ctx, tx, sendState)
 	if err != nil {
@@ -484,16 +510,14 @@ func (m *SimpleTxManager) increaseGasPrice(ctx context.Context, tx *types.Transa
 	}
 	bumpedTip, bumpedFee := updateFees(tx.GasTipCap(), tx.GasFeeCap(), tip, basefee, m.l)
 
-	// Make sure increase is at most 5x the suggested values
-	maxTip := new(big.Int).Mul(tip, big.NewInt(feeLimitMultiplier))
+	// Make sure increase is at most [FeeLimitMultiplier] the suggested values
+	maxTip := new(big.Int).Mul(tip, big.NewInt(int64(m.cfg.FeeLimitMultiplier)))
 	if bumpedTip.Cmp(maxTip) > 0 {
-		m.l.Warn(fmt.Sprintf("bumped tip getting capped at %dx multiple of the suggested value", feeLimitMultiplier), "bumped", bumpedTip, "suggestion", tip)
-		bumpedTip.Set(maxTip)
+		return nil, fmt.Errorf("bumped tip 0x%s is over %dx multiple of the suggested value", bumpedTip.Text(16), m.cfg.FeeLimitMultiplier)
 	}
-	maxFee := calcGasFeeCap(new(big.Int).Mul(basefee, big.NewInt(feeLimitMultiplier)), maxTip)
+	maxFee := calcGasFeeCap(new(big.Int).Mul(basefee, big.NewInt(int64(m.cfg.FeeLimitMultiplier))), maxTip)
 	if bumpedFee.Cmp(maxFee) > 0 {
-		m.l.Warn("bumped fee getting capped at multiple of the implied suggested value", "bumped", bumpedFee, "suggestion", maxFee)
-		bumpedFee.Set(maxFee)
+		return nil, fmt.Errorf("bumped fee 0x%s is over %dx multiple of the suggested value", bumpedFee.Text(16), m.cfg.FeeLimitMultiplier)
 	}
 	rawTx := &types.DynamicFeeTx{
 		ChainID:    tx.ChainId(),

op-service/txmgr/txmgr_test.go

@@ -80,6 +80,7 @@ func configWithNumConfs(numConfirmations uint64) Config {
 		ReceiptQueryInterval:      50 * time.Millisecond,
 		NumConfirmations:          numConfirmations,
 		SafeAbortNonceTooLowCount: 3,
+		FeeLimitMultiplier:        5,
 		TxNotInMempoolTimeout:     1 * time.Hour,
 		Signer: func(ctx context.Context, from common.Address, tx *types.Transaction) (*types.Transaction, error) {
 			return tx, nil
@@ -766,6 +767,7 @@ func doGasPriceIncrease(t *testing.T, txTipCap, txFeeCap, newTip, newBaseFee int
 			ReceiptQueryInterval:      50 * time.Millisecond,
 			NumConfirmations:          1,
 			SafeAbortNonceTooLowCount: 3,
+			FeeLimitMultiplier:        5,
 			Signer: func(ctx context.Context, from common.Address, tx *types.Transaction) (*types.Transaction, error) {
 				return tx, nil
 			},
@@ -867,6 +869,7 @@ func TestIncreaseGasPriceNotExponential(t *testing.T) {
 			ReceiptQueryInterval:      50 * time.Millisecond,
 			NumConfirmations:          1,
 			SafeAbortNonceTooLowCount: 3,
+			FeeLimitMultiplier:        5,
 			Signer: func(ctx context.Context, from common.Address, tx *types.Transaction) (*types.Transaction, error) {
 				return tx, nil
 			},
@@ -883,23 +886,20 @@ func TestIncreaseGasPriceNotExponential(t *testing.T) {
 	})
 
 	// Run IncreaseGasPrice a bunch of times in a row to simulate a very fast resubmit loop.
-	var err error
-	for i := 0; i < 30; i++ {
-		ctx := context.Background()
-		tx, err = mgr.increaseGasPrice(ctx, tx)
-		require.NoError(t, err)
+	ctx := context.Background()
+	for {
+		newTx, err := mgr.increaseGasPrice(ctx, tx)
+		if err != nil {
+			break
+		}
+		tx = newTx
 	}
 	lastTip, lastFee := tx.GasTipCap(), tx.GasFeeCap()
-	require.Equal(t, lastTip.Int64(), feeLimitMultiplier*borkedTip)
-	require.Equal(t, lastFee.Int64(), feeLimitMultiplier*(borkedTip+2*borkedFee))
+	require.Equal(t, lastTip.Int64(), int64(36))
+	require.Equal(t, lastFee.Int64(), int64(493))
 	// Confirm that fees stop rising
-	for i := 0; i < 5; i++ {
-		ctx := context.Background()
-		tx, err := mgr.increaseGasPrice(ctx, tx)
-		require.NoError(t, err)
-		require.True(t, tx.GasTipCap().Cmp(lastTip) == 0, "suggested tx tip must stop increasing")
-		require.True(t, tx.GasFeeCap().Cmp(lastFee) == 0, "suggested tx fee must stop increasing")
-	}
+	_, err := mgr.increaseGasPrice(ctx, tx)
+	require.Error(t, err)
 }
 
 func TestErrStringMatch(t *testing.T) {