op-node: implement RPC poll interval and rate-limit options and provide L1 RPC CLI options

go.mod

@@ -34,6 +34,7 @@ require (
 	golang.org/x/crypto v0.6.0
 	golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb
 	golang.org/x/term v0.5.0
+	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af
 )
 
 require (
@@ -178,7 +179,6 @@ require (
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
-	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
 	golang.org/x/tools v0.6.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect

op-e2e/migration_test.go

@@ -270,9 +270,12 @@ func TestMigration(t *testing.T) {
 	snapLog.SetHandler(log.DiscardHandler())
 	rollupNodeConfig := &node.Config{
 		L1: &node.L1EndpointConfig{
-			L1NodeAddr: forkedL1URL,
-			L1TrustRPC: false,
-			L1RPCKind:  sources.RPCKindBasic,
+			L1NodeAddr:       forkedL1URL,
+			L1TrustRPC:       false,
+			L1RPCKind:        sources.RPCKindBasic,
+			RateLimit:        0,
+			BatchSize:        20,
+			HttpPollInterval: 12 * time.Second,
 		},
 		L2: &node.L2EndpointConfig{
 			L2EngineAddr:      gethNode.HTTPAuthEndpoint(),

op-e2e/op_geth.go

@@ -75,7 +75,7 @@ func NewOpGeth(t *testing.T, ctx context.Context, cfg *SystemConfig) (*OpGeth, e
 	require.Nil(t, node.Start())
 
 	auth := rpc.WithHTTPAuth(gn.NewJWTAuth(cfg.JWTSecret))
-	l2Node, err := client.NewRPC(ctx, logger, node.WSAuthEndpoint(), auth)
+	l2Node, err := client.NewRPC(ctx, logger, node.WSAuthEndpoint(), client.WithGethRPCOptions(auth))
 	require.Nil(t, err)
 
 	// Finally create the engine client

op-e2e/setup.go

@@ -413,9 +413,12 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 			l2EndpointConfig = sys.Nodes[name].HTTPAuthEndpoint()
 		}
 		rollupCfg.L1 = &rollupNode.L1EndpointConfig{
-			L1NodeAddr: l1EndpointConfig,
-			L1TrustRPC: false,
-			L1RPCKind:  sources.RPCKindBasic,
+			L1NodeAddr:       l1EndpointConfig,
+			L1TrustRPC:       false,
+			L1RPCKind:        sources.RPCKindBasic,
+			RateLimit:        0,
+			BatchSize:        20,
+			HttpPollInterval: time.Duration(cfg.DeployConfig.L1BlockTime) * time.Second / 10,
 		}
 		rollupCfg.L2 = &rollupNode.L2EndpointConfig{
 			L2EngineAddr:      l2EndpointConfig,

op-node/client/polling.go

@@ -57,7 +57,7 @@ func NewPollingClient(ctx context.Context, lgr log.Logger, c RPC, opts ...Wrappe
 	res := &PollingClient{
 		c:         c,
 		lgr:       lgr,
-		pollRate:  250 * time.Millisecond,
+		pollRate:  12 * time.Second,
 		ctx:       ctx,
 		cancel:    cancel,
 		pollReqCh: make(chan struct{}, 1),

op-node/client/rate_limited.go

@@ -0,0 +1,52 @@
+package client
+
+import (
+	"context"
+	"time"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/rpc"
+	"golang.org/x/time/rate"
+)
+
+// RateLimitingClient is a wrapper around a pure RPC that implements a global rate-limit on requests.
+type RateLimitingClient struct {
+	c  RPC
+	rl *rate.Limiter
+}
+
+// NewRateLimitingClient implements a global rate-limit for all RPC requests.
+// A limit of N will ensure that over a long enough time-frame the given number of tokens per second is targeted.
+// Burst limits how far off we can be from the target, by specifying how many requests are allowed at once.
+func NewRateLimitingClient(c RPC, limit rate.Limit, burst int) *RateLimitingClient {
+	return &RateLimitingClient{c: c, rl: rate.NewLimiter(limit, burst)}
+}
+
+func (b *RateLimitingClient) Close() {
+	b.c.Close()
+}
+
+func (b *RateLimitingClient) CallContext(ctx context.Context, result any, method string, args ...any) error {
+	if err := b.rl.Wait(ctx); err != nil {
+		return err
+	}
+	cCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	return b.c.CallContext(cCtx, result, method, args...)
+}
+
+func (b *RateLimitingClient) BatchCallContext(ctx context.Context, batch []rpc.BatchElem) error {
+	if err := b.rl.WaitN(ctx, len(batch)); err != nil {
+		return err
+	}
+	cCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	return b.c.BatchCallContext(cCtx, batch)
+}
+
+func (b *RateLimitingClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+	if err := b.rl.Wait(ctx); err != nil {
+		return nil, err
+	}
+	return b.c.EthSubscribe(ctx, channel, args...)
+}

op-node/client/rpc.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/time/rate"
 
 	"github.com/ethereum-optimism/optimism/op-node/metrics"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -24,27 +25,85 @@ type RPC interface {
 	EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error)
 }
 
+type rpcConfig struct {
+	gethRPCOptions   []rpc.ClientOption
+	httpPollInterval time.Duration
+	backoffAttempts  int
+	limit            float64
+	burst            int
+}
+
+type RPCOption func(cfg *rpcConfig) error
+
+// WithDialBackoff configures the number of attempts for the initial dial to the RPC,
+// attempts are executed with an exponential backoff strategy.
+func WithDialBackoff(attempts int) RPCOption {
+	return func(cfg *rpcConfig) error {
+		cfg.backoffAttempts = attempts
+		return nil
+	}
+}
+
+// WithHttpPollInterval configures the RPC to poll at the given rate, in case RPC subscriptions are not available.
+func WithHttpPollInterval(duration time.Duration) RPCOption {
+	return func(cfg *rpcConfig) error {
+		cfg.httpPollInterval = duration
+		return nil
+	}
+}
+
+// WithGethRPCOptions passes the list of go-ethereum RPC options to the internal RPC instance.
+func WithGethRPCOptions(gethRPCOptions ...rpc.ClientOption) RPCOption {
+	return func(cfg *rpcConfig) error {
+		cfg.gethRPCOptions = append(cfg.gethRPCOptions, gethRPCOptions...)
+		return nil
+	}
+}
+
+// WithRateLimit configures the RPC to target the given rate limit (in requests / second).
+// See NewRateLimitingClient for more details.
+func WithRateLimit(rateLimit float64, burst int) RPCOption {
+	return func(cfg *rpcConfig) error {
+		cfg.limit = rateLimit
+		cfg.burst = burst
+		return nil
+	}
+}
+
 // NewRPC returns the correct client.RPC instance for a given RPC url.
-func NewRPC(ctx context.Context, lgr log.Logger, addr string, opts ...rpc.ClientOption) (RPC, error) {
-	underlying, err := DialRPCClientWithBackoff(ctx, lgr, addr, opts...)
+func NewRPC(ctx context.Context, lgr log.Logger, addr string, opts ...RPCOption) (RPC, error) {
+	var cfg rpcConfig
+	for i, opt := range opts {
+		if err := opt(&cfg); err != nil {
+			return nil, fmt.Errorf("rpc option %d failed to apply to RPC config: %w", i, err)
+		}
+	}
+	if cfg.backoffAttempts < 1 { // default to at least 1 attempt, or it always fails to dial.
+		cfg.backoffAttempts = 1
+	}
+	underlying, err := dialRPCClientWithBackoff(ctx, lgr, addr, cfg.backoffAttempts, cfg.gethRPCOptions...)
 	if err != nil {
 		return nil, err
 	}
 
-	wrapped := &BaseRPCClient{
-		c: underlying,
+	var wrapped RPC = &BaseRPCClient{c: underlying}
+
+	if cfg.limit != 0 {
+		wrapped = NewRateLimitingClient(wrapped, rate.Limit(cfg.limit), cfg.burst)
 	}
+
 	if httpRegex.MatchString(addr) {
-		return NewPollingClient(ctx, lgr, wrapped), nil
+		wrapped = NewPollingClient(ctx, lgr, wrapped, WithPollRate(cfg.httpPollInterval))
 	}
+
 	return wrapped, nil
 }
 
 // Dials a JSON-RPC endpoint repeatedly, with a backoff, until a client connection is established. Auth is optional.
-func DialRPCClientWithBackoff(ctx context.Context, log log.Logger, addr string, opts ...rpc.ClientOption) (*rpc.Client, error) {
+func dialRPCClientWithBackoff(ctx context.Context, log log.Logger, addr string, attempts int, opts ...rpc.ClientOption) (*rpc.Client, error) {
 	bOff := backoff.Exponential()
 	var ret *rpc.Client
-	err := backoff.DoCtx(ctx, 10, bOff, func() error {
+	err := backoff.DoCtx(ctx, attempts, bOff, func() error {
 		client, err := rpc.DialOptions(ctx, addr, opts...)
 		if err != nil {
 			if client == nil {

op-node/flags/flags.go

@@ -75,6 +75,24 @@ var (
 			return &out
 		}(),
 	}
+	L1RPCRateLimit = cli.Float64Flag{
+		Name:   "l1.rpc-rate-limit",
+		Usage:  "Optional self-imposed global rate-limit on L1 RPC requests, specified in requests / second. Disabled if set to 0.",
+		EnvVar: prefixEnvVar("L1_RPC_RATE_LIMIT"),
+		Value:  0,
+	}
+	L1RPCMaxBatchSize = cli.IntFlag{
+		Name:   "l1.rpc-max-batch-size",
+		Usage:  "Maximum number of RPC requests to bundle, e.g. during L1 blocks receipt fetching. The L1 RPC rate limit counts this as N items, but allows it to burst at once.",
+		EnvVar: prefixEnvVar("L1_RPC_MAX_BATCH_SIZE"),
+		Value:  20,
+	}
+	L1HTTPPollInterval = cli.DurationFlag{
+		Name:   "l1.http-poll-interval",
+		Usage:  "Polling interval for latest-block subscription when using an HTTP RPC provider. Ignored for other types of RPC endpoints.",
+		EnvVar: prefixEnvVar("L1_HTTP_POLL_INTERVAL"),
+		Value:  time.Second * 12,
+	}
 	L2EngineJWTSecret = cli.StringFlag{
 		Name:        "l2.jwt-secret",
 		Usage:       "Path to JWT secret key. Keys are 32 bytes, hex encoded in a file. A new key will be generated if left empty.",
@@ -196,6 +214,9 @@ var optionalFlags = []cli.Flag{
 	Network,
 	L1TrustRPC,
 	L1RPCProviderKind,
+	L1RPCRateLimit,
+	L1RPCMaxBatchSize,
+	L1HTTPPollInterval,
 	L2EngineJWTSecret,
 	VerifierL1Confs,
 	SequencerEnabledFlag,