internal/sequencerapi,miner: move conditional rate limiter to the rpc layer instead of miner

cmd/geth/main.go

@@ -155,8 +155,8 @@ var (
 		utils.GpoIgnoreGasPriceFlag,
 		utils.GpoMinSuggestedPriorityFeeFlag,
 		utils.RollupSequencerHTTPFlag,
-		utils.RollupSequencerEnableTxConditionalFlag,
-		utils.RollupSequencerTxConditionalRateLimitFlag,
+		utils.RollupSequencerTxConditionalEnabledFlag,
+		utils.RollupSequencerTxConditionalCostRateLimitFlag,
 		utils.RollupHistoricalRPCFlag,
 		utils.RollupHistoricalRPCTimeoutFlag,
 		utils.RollupDisableTxPoolGossipFlag,

cmd/utils/flags.go

@@ -976,14 +976,14 @@ var (
 		Category: flags.RollupCategory,
 		Value:    true,
 	}
-	RollupSequencerEnableTxConditionalFlag = &cli.BoolFlag{
-		Name:     "rollup.sequencerenabletxconditional",
+	RollupSequencerTxConditionalEnabledFlag = &cli.BoolFlag{
+		Name:     "rollup.sequencertxconditionalenabled",
 		Usage:    "Serve the eth_sendRawTransactionConditional endpoint and apply the conditional constraints on mempool inclusion & block building",
 		Category: flags.RollupCategory,
 		Value:    false,
 	}
-	RollupSequencerTxConditionalRateLimitFlag = &cli.IntFlag{
-		Name:     "rollup.sequencertxconditionalratelimit",
+	RollupSequencerTxConditionalCostRateLimitFlag = &cli.IntFlag{
+		Name:     "rollup.sequencertxconditionalcostratelimit",
 		Usage:    "Maximum cost -- storage lookups -- allowed for conditional transactions in a given second",
 		Category: flags.RollupCategory,
 		Value:    5000,
@@ -1729,9 +1729,6 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
 	if ctx.IsSet(RollupComputePendingBlock.Name) {
 		cfg.RollupComputePendingBlock = ctx.Bool(RollupComputePendingBlock.Name)
 	}
-
-	// This flag has a default rate limit so always set
-	cfg.RollupTransactionConditionalRateLimit = ctx.Int(RollupSequencerTxConditionalRateLimitFlag.Name)
 }
 
 func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {
@@ -1970,7 +1967,8 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	cfg.RollupDisableTxPoolAdmission = cfg.RollupSequencerHTTP != "" && !ctx.Bool(RollupEnableTxPoolAdmissionFlag.Name)
 	cfg.RollupHaltOnIncompatibleProtocolVersion = ctx.String(RollupHaltOnIncompatibleProtocolVersionFlag.Name)
 	cfg.ApplySuperchainUpgrades = ctx.Bool(RollupSuperchainUpgradesFlag.Name)
-	cfg.RollupSequencerEnableTxConditional = ctx.Bool(RollupSequencerEnableTxConditionalFlag.Name)
+	cfg.RollupSequencerTxConditionalEnabled = ctx.Bool(RollupSequencerTxConditionalEnabledFlag.Name)
+	cfg.RollupSequencerTxConditionalCostRateLimit = ctx.Int(RollupSequencerTxConditionalCostRateLimitFlag.Name)
 
 	// Override any default configs for hard coded networks.
 	switch {

eth/backend.go

@@ -59,6 +59,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
+	"golang.org/x/time/rate"
 )
 
 // Config contains the configuration options of the ETH protocol.
@@ -376,9 +377,10 @@ func (s *Ethereum) APIs() []rpc.API {
 	apis = append(apis, s.engine.APIs(s.BlockChain())...)
 
 	// Append any Sequencer APIs as enabled
-	if s.config.RollupSequencerEnableTxConditional {
+	if s.config.RollupSequencerTxConditionalEnabled {
 		log.Info("Enabling eth_sendRawTransactionConditional endpoint support")
-		apis = append(apis, sequencerapi.GetSendRawTxConditionalAPI(s.APIBackend, s.seqRPCService))
+		costRateLimit := rate.Limit(s.config.RollupSequencerTxConditionalCostRateLimit)
+		apis = append(apis, sequencerapi.GetSendRawTxConditionalAPI(s.APIBackend, s.seqRPCService, costRateLimit))
 	}
 
 	// Append all the local APIs and return

eth/ethconfig/config.go

@@ -183,13 +183,14 @@ type Config struct {
 	// ApplySuperchainUpgrades requests the node to load chain-configuration from the superchain-registry.
 	ApplySuperchainUpgrades bool `toml:",omitempty"`
 
-	RollupSequencerHTTP                     string
-	RollupSequencerEnableTxConditional      bool
-	RollupHistoricalRPC                     string
-	RollupHistoricalRPCTimeout              time.Duration
-	RollupDisableTxPoolGossip               bool
-	RollupDisableTxPoolAdmission            bool
-	RollupHaltOnIncompatibleProtocolVersion string
+	RollupSequencerHTTP                       string
+	RollupSequencerTxConditionalEnabled       bool
+	RollupSequencerTxConditionalCostRateLimit int
+	RollupHistoricalRPC                       string
+	RollupHistoricalRPCTimeout                time.Duration
+	RollupDisableTxPoolGossip                 bool
+	RollupDisableTxPoolAdmission              bool
+	RollupHaltOnIncompatibleProtocolVersion   string
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain config.

internal/sequencerapi/api.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -22,14 +23,18 @@ var (
 )
 
 type sendRawTxCond struct {
-	b      ethapi.Backend
-	seqRPC *rpc.Client
+	b           ethapi.Backend
+	seqRPC      *rpc.Client
+	costLimiter *rate.Limiter
 }
 
-func GetSendRawTxConditionalAPI(b ethapi.Backend, seqRPC *rpc.Client) rpc.API {
+func GetSendRawTxConditionalAPI(b ethapi.Backend, seqRPC *rpc.Client, costRateLimit rate.Limit) rpc.API {
+	// Applying a manual bump to the burst to allow conditional txs to queue. Metrics will
+	// will inform of adjustments that may need to be made here.
+	costLimiter := rate.NewLimiter(costRateLimit, 3*params.TransactionConditionalMaxCost)
 	return rpc.API{
 		Namespace: "eth",
-		Service:   &sendRawTxCond{b, seqRPC},
+		Service:   &sendRawTxCond{b, seqRPC, costLimiter},
 	}
 }
 
@@ -83,6 +88,14 @@ func (s *sendRawTxCond) SendRawTransactionConditional(ctx context.Context, txByt
 		}
 	}
 
+	// enforce rate limit on the cost to be observed
+	if err := s.costLimiter.WaitN(ctx, cost); err != nil {
+		return common.Hash{}, &rpc.JsonError{
+			Message: fmt.Sprintf("cost %d rate limited", cost),
+			Code:    params.TransactionConditionalCostExceededMaxErrCode,
+		}
+	}
+
 	tx := new(types.Transaction)
 	if err := tx.UnmarshalBinary(txBytes); err != nil {
 		return common.Hash{}, err

miner/miner.go

@@ -33,7 +33,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/tracers"
 	"github.com/ethereum/go-ethereum/params"
-	"golang.org/x/time/rate"
 )
 
 // Backend wraps all methods required for mining. Only full node is capable
@@ -87,9 +86,6 @@ type Miner struct {
 	pendingMu   sync.Mutex // Lock protects the pending block
 
 	backend Backend
-
-	// TransactionConditional safegaurds
-	conditionalLimiter *rate.Limiter
 }
 
 // New creates a new miner with provided config.
@@ -102,8 +98,6 @@ func New(eth Backend, config Config, engine consensus.Engine) *Miner {
 		txpool:      eth.TxPool(),
 		chain:       eth.BlockChain(),
 		pending:     &pending{},
-		// setup the rate limit imposed on conditional transactions when block building
-		conditionalLimiter: rate.NewLimiter(rate.Limit(config.RollupTransactionConditionalRateLimit), params.TransactionConditionalMaxCost),
 	}
 }
 

miner/worker.go

@@ -47,8 +47,7 @@ const (
 )
 
 var (
-	errTxConditionalInvalid     = errors.New("transaction conditional failed")
-	errTxConditionalRateLimited = errors.New("transaction conditional rate limited")
+	errTxConditionalInvalid = errors.New("transaction conditional failed")
 
 	errBlockInterruptedByNewHead  = errors.New("new head arrived while building block")
 	errBlockInterruptedByRecommit = errors.New("recommit interrupt while building block")
@@ -297,16 +296,6 @@ func (miner *Miner) commitTransaction(env *environment, tx *types.Transaction) e
 
 	// If a conditional is set, check prior to applying
 	if conditional := tx.Conditional(); conditional != nil {
-		now, cost := time.Now(), conditional.Cost()
-		res := miner.conditionalLimiter.ReserveN(now, cost)
-		if !res.OK() {
-			return fmt.Errorf("exceeded rate limiter burst: cost %d, burst %d: %w", cost, miner.conditionalLimiter.Burst(), errTxConditionalInvalid)
-		}
-		if res.Delay() > 0 {
-			res.Cancel()
-			return fmt.Errorf("exceeded rate limit: cost %d, available tokens %f: %w", cost, miner.conditionalLimiter.Tokens(), errTxConditionalRateLimited)
-		}
-
 		txConditionalMinedTimer.UpdateSince(tx.Time())
 
 		// check the conditional
@@ -462,14 +451,6 @@ func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *tran
 			log.Warn("Skipping account, transaction with failed conditional", "sender", from, "hash", ltx.Hash, "err", err)
 			txs.Pop()
 
-		case errors.Is(err, errTxConditionalRateLimited):
-			// err contains contextual info of the cost and limiter tokens available
-			txConditionalRejectedCounter.Inc(1)
-
-			// note: we do not mark the tx as rejected as it is still eligible for inclusion at a later time
-			log.Warn("Skipping account, transaction with conditional rate limited", "sender", from, "hash", ltx.Hash, "err", err)
-			txs.Pop()
-
 		case errors.Is(err, nil):
 			// Everything ok, collect the logs and shift in the next transaction from the same account
 			txs.Shift()