op-batcher: optimize tx submitting and add metrics (#195)

Co-authored-by: Sebastian Stammler <[email protected]>

op-batcher/metrics/metrics.go

@@ -185,8 +185,8 @@ func NewMetrics(procName string) *Metrics {
 		blobUsedBytes: factory.NewHistogram(prometheus.HistogramOpts{
 			Namespace: ns,
 			Name:      "blob_used_bytes",
-			Help:      "Blob size in bytes being submitted.",
-			Buckets:   prometheus.LinearBuckets(0.0, eth.MaxBlobDataSize/13, 13),
+			Help:      "Blob size in bytes (of last blob only for multi-blob txs).",
+			Buckets:   prometheus.LinearBuckets(0.0, eth.MaxBlobDataSize/13, 14),
 		}),
 
 		batcherTxEvs: opmetrics.NewEventVec(factory, ns, "", "batcher_tx", "BatcherTx", []string{"stage"}),

op-batcher/metrics/noop.go

@@ -45,5 +45,7 @@ func (*noopMetrics) RecordBlobUsedBytes(int) {}
 func (*noopMetrics) StartBalanceMetrics(log.Logger, ethereum.ChainStateReader, common.Address) io.Closer {
 	return nil
 }
+func (*noopMetrics) RecordBlobsNumber(number int) {}
+
 func (m *noopMetrics) RecordL1UrlSwitchEvt(url string) {
 }

op-proposer/metrics/noop.go

@@ -27,6 +27,7 @@ func (*noopMetrics) RecordL2BlocksProposed(l2ref eth.L2BlockRef) {}
 func (*noopMetrics) StartBalanceMetrics(log.Logger, ethereum.ChainStateReader, common.Address) io.Closer {
 	return nil
 }
+func (*noopMetrics) RecordBlobsNumber(number int) {}
 
 func (m *noopMetrics) RecordL1UrlSwitchEvt(url string) {
 }

op-service/txmgr/cli.go

@@ -31,6 +31,7 @@ const (
 	SafeAbortNonceTooLowCountFlagName = "safe-abort-nonce-too-low-count"
 	FeeLimitMultiplierFlagName        = "fee-limit-multiplier"
 	FeeLimitThresholdFlagName         = "txmgr.fee-limit-threshold"
+	BlobGasPriceLimitFlagName         = "txmgr.blob-gas-price-limit"
 	MinBaseFeeFlagName                = "txmgr.min-basefee"
 	MinTipCapFlagName                 = "txmgr.min-tip-cap"
 	ResubmissionTimeoutFlagName       = "resubmission-timeout"
@@ -60,6 +61,7 @@ type DefaultFlagValues struct {
 	SafeAbortNonceTooLowCount uint64
 	FeeLimitMultiplier        uint64
 	FeeLimitThresholdGwei     float64
+	BlobGasPriceLimitGwei     float64
 	MinTipCapGwei             float64
 	MinBaseFeeGwei            float64
 	ResubmissionTimeout       time.Duration
@@ -75,6 +77,7 @@ var (
 		SafeAbortNonceTooLowCount: uint64(3),
 		FeeLimitMultiplier:        uint64(5),
 		FeeLimitThresholdGwei:     100.0,
+		BlobGasPriceLimitGwei:     100.0,
 		MinTipCapGwei:             1.0,
 		MinBaseFeeGwei:            1.0,
 		ResubmissionTimeout:       48 * time.Second,
@@ -88,6 +91,7 @@ var (
 		SafeAbortNonceTooLowCount: uint64(3),
 		FeeLimitMultiplier:        uint64(5),
 		FeeLimitThresholdGwei:     100.0,
+		BlobGasPriceLimitGwei:     100.0,
 		MinTipCapGwei:             1.0,
 		MinBaseFeeGwei:            1.0,
 		ResubmissionTimeout:       24 * time.Second,
@@ -146,6 +150,12 @@ func CLIFlagsWithDefaults(envPrefix string, defaults DefaultFlagValues) []cli.Fl
 			Value:   defaults.FeeLimitThresholdGwei,
 			EnvVars: prefixEnvVars("TXMGR_FEE_LIMIT_THRESHOLD"),
 		},
+		&cli.Float64Flag{
+			Name:    BlobGasPriceLimitFlagName,
+			Usage:   "The maximum limit (in GWei) of blob gas price, above which will stop submit and wait for the price go down",
+			Value:   defaults.BlobGasPriceLimitGwei,
+			EnvVars: prefixEnvVars("TXMGR_BLOB_GAS_PRICE_LIMIT"),
+		},
 		&cli.Float64Flag{
 			Name:    MinTipCapFlagName,
 			Usage:   "Enforces a minimum tip cap (in GWei) to use when determining tx fees. 1 GWei by default.",
@@ -203,6 +213,7 @@ type CLIConfig struct {
 	SafeAbortNonceTooLowCount uint64
 	FeeLimitMultiplier        uint64
 	FeeLimitThresholdGwei     float64
+	BlobGasPriceLimitGwei     float64
 	MinBaseFeeGwei            float64
 	MinTipCapGwei             float64
 	ResubmissionTimeout       time.Duration
@@ -219,6 +230,7 @@ func NewCLIConfig(l1RPCURL string, defaults DefaultFlagValues) CLIConfig {
 		SafeAbortNonceTooLowCount: defaults.SafeAbortNonceTooLowCount,
 		FeeLimitMultiplier:        defaults.FeeLimitMultiplier,
 		FeeLimitThresholdGwei:     defaults.FeeLimitThresholdGwei,
+		BlobGasPriceLimitGwei:     defaults.BlobGasPriceLimitGwei,
 		MinTipCapGwei:             defaults.MinTipCapGwei,
 		MinBaseFeeGwei:            defaults.MinBaseFeeGwei,
 		ResubmissionTimeout:       defaults.ResubmissionTimeout,
@@ -278,6 +290,7 @@ func ReadCLIConfig(ctx *cli.Context) CLIConfig {
 		SafeAbortNonceTooLowCount: ctx.Uint64(SafeAbortNonceTooLowCountFlagName),
 		FeeLimitMultiplier:        ctx.Uint64(FeeLimitMultiplierFlagName),
 		FeeLimitThresholdGwei:     ctx.Float64(FeeLimitThresholdFlagName),
+		BlobGasPriceLimitGwei:     ctx.Float64(BlobGasPriceLimitFlagName),
 		MinBaseFeeGwei:            ctx.Float64(MinBaseFeeFlagName),
 		MinTipCapGwei:             ctx.Float64(MinTipCapFlagName),
 		ResubmissionTimeout:       ctx.Duration(ResubmissionTimeoutFlagName),
@@ -326,6 +339,11 @@ func NewConfig(cfg CLIConfig, l log.Logger, m txmetrics.TxMetricer) (Config, err
 		return Config{}, fmt.Errorf("invalid fee limit threshold: %w", err)
 	}
 
+	blobGasPriceLimit, err := eth.GweiToWei(cfg.BlobGasPriceLimitGwei)
+	if err != nil {
+		return Config{}, fmt.Errorf("invalid blob gas price limit: %w", err)
+	}
+
 	minBaseFee, err := eth.GweiToWei(cfg.MinBaseFeeGwei)
 	if err != nil {
 		return Config{}, fmt.Errorf("invalid min base fee: %w", err)
@@ -341,6 +359,7 @@ func NewConfig(cfg CLIConfig, l log.Logger, m txmetrics.TxMetricer) (Config, err
 		ResubmissionTimeout:       cfg.ResubmissionTimeout,
 		FeeLimitMultiplier:        cfg.FeeLimitMultiplier,
 		FeeLimitThreshold:         feeLimitThreshold,
+		BlobGasPriceLimit:         blobGasPriceLimit,
 		MinBaseFee:                minBaseFee,
 		MinTipCap:                 minTipCap,
 		ChainID:                   chainID,
@@ -372,6 +391,10 @@ type Config struct {
 	// below this threshold.
 	FeeLimitThreshold *big.Int
 
+	// The maximum limit (in GWei) of blob gas price,
+	// Above which will stop submit and wait for the price go down
+	BlobGasPriceLimit *big.Int
+
 	// Minimum base fee (in Wei) to assume when determining tx fees.
 	MinBaseFee *big.Int
 

op-service/txmgr/metrics/noop.go

@@ -19,6 +19,7 @@ func (*NoopTxMetrics) RecordTxConfirmationLatency(int64) {}
 func (*NoopTxMetrics) TxConfirmed(*types.Receipt)        {}
 func (*NoopTxMetrics) TxPublished(string)                {}
 func (*NoopTxMetrics) RecordBaseFee(*big.Int)            {}
+func (*NoopTxMetrics) RecordBlobBaseFee(*big.Int)        {}
 func (*NoopTxMetrics) RecordTipCap(*big.Int)             {}
 func (*NoopTxMetrics) RPCError()                         {}
 func (m *NoopTxMetrics) RecordL1UrlSwitchEvt(url string) {}

op-service/txmgr/metrics/tx_metrics.go

@@ -18,6 +18,8 @@ type TxMetricer interface {
 	TxConfirmed(*types.Receipt)
 	TxPublished(string)
 	RecordBaseFee(*big.Int)
+	RecordBlobBaseFee(*big.Int)
+	RecordBlobsNumber(int)
 	RecordTipCap(*big.Int)
 	RPCError()
 	client.FallbackClientMetricer
@@ -37,6 +39,8 @@ type TxMetrics struct {
 	publishEvent       *metrics.Event
 	confirmEvent       metrics.EventVec
 	baseFee            prometheus.Gauge
+	blobBaseFee        prometheus.Gauge
+	blobsNumber        prometheus.Gauge
 	tipCap             prometheus.Gauge
 	rpcError           prometheus.Counter
 	*client.FallbackClientMetrics
@@ -112,7 +116,19 @@ func MakeTxMetrics(ns string, factory metrics.Factory) TxMetrics {
 		baseFee: factory.NewGauge(prometheus.GaugeOpts{
 			Namespace: ns,
 			Name:      "basefee_wei",
-			Help:      "Latest L1 baseFee (in Wei)",
+			Help:      "Latest L1 base fee (in Wei)",
+			Subsystem: "txmgr",
+		}),
+		blobBaseFee: factory.NewGauge(prometheus.GaugeOpts{
+			Namespace: ns,
+			Name:      "blob_basefee_wei",
+			Help:      "Latest Blob base fee (in Wei)",
+			Subsystem: "txmgr",
+		}),
+		blobsNumber: factory.NewGauge(prometheus.GaugeOpts{
+			Namespace: ns,
+			Name:      "blobs_number_in_tx",
+			Help:      "number of blobs in tx",
 			Subsystem: "txmgr",
 		}),
 		tipCap: factory.NewGauge(prometheus.GaugeOpts{
@@ -169,6 +185,15 @@ func (t *TxMetrics) RecordBaseFee(baseFee *big.Int) {
 	t.baseFee.Set(bff)
 }
 
+func (t *TxMetrics) RecordBlobBaseFee(blobBaseFee *big.Int) {
+	bff, _ := blobBaseFee.Float64()
+	t.blobBaseFee.Set(bff)
+}
+
+func (t *TxMetrics) RecordBlobsNumber(number int) {
+	t.blobsNumber.Set(float64(number))
+}
+
 func (t *TxMetrics) RecordTipCap(tipcap *big.Int) {
 	tcf, _ := tipcap.Float64()
 	t.tipCap.Set(tcf)

op-service/txmgr/txmgr.go

@@ -293,7 +293,9 @@ func (m *SimpleTxManager) craftTx(ctx context.Context, candidate TxCandidate) (*
 		if blobBaseFee == nil {
 			return nil, fmt.Errorf("expected non-nil blobBaseFee")
 		}
-		blobFeeCap := calcBlobFeeCap(blobBaseFee)
+		// no need to calcBlobFeeCap, prefer raw blobBaseFee
+		// blobFeeCap := calcBlobFeeCap(blobBaseFee)
+		blobFeeCap := blobBaseFee
 		message := &types.BlobTx{
 			To:         *candidate.To,
 			Data:       candidate.TxData,
@@ -317,7 +319,6 @@ func (m *SimpleTxManager) craftTx(ctx context.Context, candidate TxCandidate) (*
 		}
 	}
 	return m.signWithNextNonce(ctx, txMessage) // signer sets the nonce field of the tx
-
 }
 
 // MakeSidecar builds & returns the BlobTxSidecar and corresponding blob hashes from the raw blob
@@ -507,6 +508,12 @@ func (m *SimpleTxManager) publishTx(ctx context.Context, tx *types.Transaction,
 		case errStringMatch(err, core.ErrNonceTooLow):
 			l.Warn("nonce too low", "err", err)
 			m.metr.TxPublished("nonce_too_low")
+		case errStringMatch(err, core.ErrNonceTooHigh):
+			l.Warn("nonce too high", "err", err)
+			m.metr.TxPublished("nonce_too_high")
+			bumpFeesImmediately = false // retry without fee bump
+			time.Sleep(100*time.Millisecond)
+			continue
 		case errStringMatch(err, context.Canceled):
 			m.metr.RPCError()
 			l.Warn("transaction send cancelled", "err", err)
@@ -546,6 +553,9 @@ func (m *SimpleTxManager) waitForTx(ctx context.Context, tx *types.Transaction,
 	}
 	select {
 	case receiptChan <- receipt:
+		if blobHashes := tx.BlobHashes(); blobHashes != nil {
+			m.metr.RecordBlobsNumber(len(blobHashes))
+		}
 		m.metr.RecordTxConfirmationLatency(time.Since(t).Milliseconds())
 	default:
 	}
@@ -599,6 +609,11 @@ func (m *SimpleTxManager) queryReceipt(ctx context.Context, txHash common.Hash,
 	}
 
 	m.metr.RecordBaseFee(tip.BaseFee)
+	if tip.ExcessBlobGas != nil {
+		blobFee := eip4844.CalcBlobFee(*tip.ExcessBlobGas)
+		m.metr.RecordBlobBaseFee(blobFee)
+	}
+
 	m.l.Debug("Transaction mined, checking confirmations", "tx", txHash,
 		"block", eth.ReceiptBlockID(receipt), "tip", eth.HeaderBlockID(tip),
 		"numConfirmations", m.cfg.NumConfirmations)
@@ -753,6 +768,7 @@ func (m *SimpleTxManager) suggestGasPriceCaps(ctx context.Context) (*big.Int, *b
 	var blobFee *big.Int
 	if head.ExcessBlobGas != nil {
 		blobFee = eip4844.CalcBlobFee(*head.ExcessBlobGas)
+		m.metr.RecordBlobBaseFee(blobFee)
 	}
 	return tip, baseFee, blobFee, nil
 }
@@ -783,6 +799,13 @@ func (m *SimpleTxManager) checkLimits(tip, baseFee, bumpedTip, bumpedFee *big.In
 }
 
 func (m *SimpleTxManager) checkBlobFeeLimits(blobBaseFee, bumpedBlobFee *big.Int) error {
+	// If above limit, do not send transaction
+	if limit := m.cfg.BlobGasPriceLimit; limit != nil && limit.Cmp(bumpedBlobFee) == -1 {
+		return fmt.Errorf(
+			"bumped blob fee %v is over blob gas price limit value: %v",
+			bumpedBlobFee, m.cfg.BlobGasPriceLimit)
+	}
+
 	// If below threshold, don't apply multiplier limit. Note we use same threshold parameter here
 	// used for non-blob fee limiting.
 	if thr := m.cfg.FeeLimitThreshold; thr != nil && thr.Cmp(bumpedBlobFee) == 1 {