Skip to content

N
nebula
Commit 753c36ca authored by Michael de Hoog's avatar Michael de Hoog
Browse files
Options

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

parent cf7b329a
Show whitespace changes
Inline Side-by-side
Showing with 123 additions and 70 deletions
op-e2e/setup.go
View file @ 753c36ca
......@@ -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
View file @ 753c36ca
......@@ -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("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
View file @ 753c36ca
......@@ -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
View file @ 753c36ca
......@@ -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,17 @@ 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)
sendTx := func(tx *types.Transaction, bumpFees bool) *types.Transaction {
return m.publishAndWaitForTx(ctx, tx, sendState, receiptChan, bumpFees, wg.Done)
}
// Immediately publish a transaction before starting the resumbission loop
wg.Add(1)
go sendTxAsync(tx)
tx = sendTx(tx, false)
ticker := time.NewTicker(m.cfg.ResubmissionTimeout)
defer ticker.Stop()
bumpCounter := 0
for {
select {
case <-ticker.C:
......@@ -322,25 +317,14 @@ 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 = sendTx(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
}
......@@ -350,45 +334,86 @@ func (m *SimpleTxManager) sendTx(ctx context.Context, tx *types.Transaction) (*t
// 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")
func (m *SimpleTxManager) publishAndWaitForTx(ctx context.Context, tx *types.Transaction, sendState *SendState, receiptChan chan *types.Receipt, bumpFees bool, done func()) *types.Transaction {
waiting := false
defer func() {
if !waiting {
done()
}
}()
cCtx, cancel := context.WithTimeout(ctx, m.cfg.NetworkTimeout)
defer cancel()
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)
l.Info("Publishing transaction")
t := time.Now()
for {
if bumpFees {
newTx, err := m.increaseGasPrice(ctx, tx)
if err != nil {
l.Error("unable to increase gas", "err", err)
m.metr.TxPublished("bump_failed")
return tx
}
tx = newTx
sendState.bumpCount++
l = updateLogFields(tx)
}
bumpFees = 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
}
cCtx, cancel := context.WithTimeout(ctx, m.cfg.NetworkTimeout)
err := m.backend.SendTransaction(cCtx, tx)
cancel()
sendState.ProcessSendError(err)
// Properly log & exit if there is an error
if err != nil {
if err == nil {
break
}
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
}
m.metr.TxPublished("")
m.metr.TxPublished("")
log.Info("Transaction successfully published")
waiting = true
go func() {
defer done()
// Poll for the transaction to be ready & then send the result to receiptChan
receipt, err := m.waitMined(ctx, tx, sendState)
if err != nil {
......@@ -401,6 +426,9 @@ func (m *SimpleTxManager) publishAndWaitForTx(ctx context.Context, tx *types.Tra
m.metr.RecordTxConfirmationLatency(time.Since(t).Milliseconds())
default:
}
}()
return tx
}
// waitMined waits for the transaction to be mined or for the context to be cancelled.
......@@ -485,15 +513,13 @@ 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))
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
View file @ 753c36ca
......@@ -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)
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) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment