Merge branch 'develop' into jg/sync_start_logging

.changeset/fresh-pots-move.md

+---
+'@eth-optimism/batch-submitter-service': patch
+---
+
+Allow deposit only batches

batch-submitter/drivers/sequencer/batch_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/ethereum-optimism/optimism/batch-submitter/drivers/sequencer"
 	l2common "github.com/ethereum-optimism/optimism/l2geth/common"
+	"github.com/ethereum-optimism/optimism/l2geth/core/types"
 	l2types "github.com/ethereum-optimism/optimism/l2geth/core/types"
 	"github.com/stretchr/testify/require"
 )
@@ -47,3 +48,76 @@ func TestBatchElementFromBlock(t *testing.T) {
 	require.False(t, element.IsSequencerTx())
 	require.Nil(t, element.Tx)
 }
+
+func TestGenSequencerParams(t *testing.T) {
+	tx := types.NewTransaction(0, l2common.Address{}, big.NewInt(0), 0, big.NewInt(0), []byte{})
+
+	shouldStartAtElement := uint64(1)
+	blockOffset := uint64(1)
+	batches := []sequencer.BatchElement{
+		{Timestamp: 1, BlockNumber: 1},
+		{Timestamp: 1, BlockNumber: 1, Tx: sequencer.NewCachedTx(tx)},
+	}
+
+	params, err := sequencer.GenSequencerBatchParams(shouldStartAtElement, blockOffset, batches)
+	require.NoError(t, err)
+
+	require.Equal(t, uint64(0), params.ShouldStartAtElement)
+	require.Equal(t, uint64(len(batches)), params.TotalElementsToAppend)
+	require.Equal(t, len(batches), len(params.Contexts))
+	// There is only 1 sequencer tx
+	require.Equal(t, 1, len(params.Txs))
+
+	// There are 2 contexts
+	// The first context contains the deposit
+	context1 := params.Contexts[0]
+	require.Equal(t, uint64(0), context1.NumSequencedTxs)
+	require.Equal(t, uint64(1), context1.NumSubsequentQueueTxs)
+	require.Equal(t, uint64(1), context1.Timestamp)
+	require.Equal(t, uint64(1), context1.BlockNumber)
+
+	// The second context contains the sequencer tx
+	context2 := params.Contexts[1]
+	require.Equal(t, uint64(1), context2.NumSequencedTxs)
+	require.Equal(t, uint64(0), context2.NumSubsequentQueueTxs)
+	require.Equal(t, uint64(1), context2.Timestamp)
+	require.Equal(t, uint64(1), context2.BlockNumber)
+}
+
+func TestGenSequencerParamsOnlyDeposits(t *testing.T) {
+	shouldStartAtElement := uint64(1)
+	blockOffset := uint64(1)
+	batches := []sequencer.BatchElement{
+		{Timestamp: 1, BlockNumber: 1},
+		{Timestamp: 1, BlockNumber: 1},
+		{Timestamp: 2, BlockNumber: 2},
+	}
+
+	params, err := sequencer.GenSequencerBatchParams(shouldStartAtElement, blockOffset, batches)
+	require.NoError(t, err)
+
+	// The batches will pack deposits into the same context when their
+	// timestamps and blocknumbers are the same
+	require.Equal(t, uint64(0), params.ShouldStartAtElement)
+	require.Equal(t, uint64(len(batches)), params.TotalElementsToAppend)
+	// 2 deposits have the same timestamp + blocknumber, they go in the
+	// same context. 1 deposit has a different timestamp + blocknumber,
+	// it goes into a different context. Therefore there are 2 contexts
+	require.Equal(t, 2, len(params.Contexts))
+	// No sequencer txs
+	require.Equal(t, 0, len(params.Txs))
+
+	// There are 2 contexts
+	// The first context contains the deposit
+	context1 := params.Contexts[0]
+	require.Equal(t, uint64(0), context1.NumSequencedTxs)
+	require.Equal(t, uint64(2), context1.NumSubsequentQueueTxs)
+	require.Equal(t, uint64(1), context1.Timestamp)
+	require.Equal(t, uint64(1), context1.BlockNumber)
+
+	context2 := params.Contexts[1]
+	require.Equal(t, uint64(0), context2.NumSequencedTxs)
+	require.Equal(t, uint64(1), context2.NumSubsequentQueueTxs)
+	require.Equal(t, uint64(2), context2.Timestamp)
+	require.Equal(t, uint64(2), context2.BlockNumber)
+}

batch-submitter/drivers/sequencer/encoding.go

@@ -222,11 +222,6 @@ func (p *AppendSequencerBatchParams) Write(
 		return ErrMalformedBatch
 	}
 
-	// There must be transactions if there are contexts
-	if len(p.Txs) == 0 && len(p.Contexts) != 0 {
-		return ErrMalformedBatch
-	}
-
 	// copy the contexts as to not malleate the struct
 	// when it is a typed batch
 	contexts := make([]BatchContext, 0, len(p.Contexts)+1)
@@ -361,9 +356,6 @@ func (p *AppendSequencerBatchParams) Read(r io.Reader) error {
 			if len(p.Contexts) == 0 && len(p.Txs) != 0 {
 				return ErrMalformedBatch
 			}
-			if len(p.Txs) == 0 && len(p.Contexts) != 0 {
-				return ErrMalformedBatch
-			}
 			return closeReader()
 		} else if err != nil {
 			return err

batch-submitter/drivers/sequencer/testdata/valid_append_sequencer_batch_params.json

@@ -46,7 +46,7 @@
         }
       ],
       "txs": [],
-      "error": true
+      "error": false
     },
     {
       "name": "multiple-contexts-no-txs",
@@ -80,7 +80,7 @@
         }
       ],
       "txs": [],
-      "error": true
+      "error": false
     },
     {
       "name": "complex",

op-batcher/metrics/metrics.go

@@ -185,7 +185,7 @@ func (m *Metrics) RecordLatestL1Block(l1ref eth.L1BlockRef) {
 	m.RecordL1Ref("latest", l1ref)
 }
 
-// RecordL2BlockLoaded should be called when a new L2 block was loaded into the
+// RecordL2BlocksLoaded should be called when a new L2 block was loaded into the
 // channel manager (but not processed yet).
 func (m *Metrics) RecordL2BlocksLoaded(l2ref eth.L2BlockRef) {
 	m.RecordL2Ref(StageLoaded, l2ref)

op-e2e/actions/l2_proposer.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum-optimism/optimism/op-node/sources"
+	"github.com/ethereum-optimism/optimism/op-proposer/metrics"
 	"github.com/ethereum-optimism/optimism/op-proposer/proposer"
 	opcrypto "github.com/ethereum-optimism/optimism/op-service/crypto"
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
@@ -60,7 +61,7 @@ func NewL2Proposer(t Testing, log log.Logger, cfg *ProposerCfg, l1 *ethclient.Cl
 		SignerFnFactory:   signer,
 	}
 
-	dr, err := proposer.NewL2OutputSubmitter(proposerCfg, log)
+	dr, err := proposer.NewL2OutputSubmitter(proposerCfg, log, metrics.NoopMetrics)
 	require.NoError(t, err)
 
 	return &L2Proposer{

op-e2e/migration_test.go

@@ -15,6 +15,7 @@ import (
 	batchermetrics "github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
+	proposermetrics "github.com/ethereum-optimism/optimism/op-proposer/metrics"
 	l2os "github.com/ethereum-optimism/optimism/op-proposer/proposer"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 
@@ -277,6 +278,7 @@ func TestMigration(t *testing.T) {
 			L2EngineAddr:      gethNode.HTTPAuthEndpoint(),
 			L2EngineJWTSecret: testingJWTSecret,
 		},
+		L2Sync: &node.PreparedL2SyncEndpoint{Client: nil, TrustRPC: false},
 		Driver: driver.Config{
 			VerifierConfDepth:  0,
 			SequencerConfDepth: 0,
@@ -362,7 +364,7 @@ func TestMigration(t *testing.T) {
 			Format: "text",
 		},
 		PrivateKey: hexPriv(secrets.Proposer),
-	}, lgr.New("module", "proposer"))
+	}, lgr.New("module", "proposer"), proposermetrics.NoopMetrics)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		proposer.Stop()

op-e2e/setup.go

@@ -37,6 +37,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/driver"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
 	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	proposermetrics "github.com/ethereum-optimism/optimism/op-proposer/metrics"
 	l2os "github.com/ethereum-optimism/optimism/op-proposer/proposer"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 )
@@ -193,6 +194,9 @@ type SystemConfig struct {
 
 	// If the proposer can make proposals for L2 blocks derived from L1 blocks which are not finalized on L1 yet.
 	NonFinalizedProposals bool
+
+	// Explicitly disable batcher, for tests that rely on unsafe L2 payloads
+	DisableBatcher bool
 }
 
 type System struct {
@@ -417,6 +421,10 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 			L2EngineAddr:      l2EndpointConfig,
 			L2EngineJWTSecret: cfg.JWTSecret,
 		}
+		rollupCfg.L2Sync = &rollupNode.PreparedL2SyncEndpoint{
+			Client:   nil,
+			TrustRPC: false,
+		}
 	}
 
 	// Geth Clients
@@ -572,7 +580,7 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 			Format: "text",
 		},
 		PrivateKey: hexPriv(cfg.Secrets.Proposer),
-	}, sys.cfg.Loggers["proposer"])
+	}, sys.cfg.Loggers["proposer"], proposermetrics.NoopMetrics)
 	if err != nil {
 		return nil, fmt.Errorf("unable to setup l2 output submitter: %w", err)
 	}
@@ -606,8 +614,11 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 		return nil, fmt.Errorf("failed to setup batch submitter: %w", err)
 	}
 
-	if err := sys.BatchSubmitter.Start(); err != nil {
-		return nil, fmt.Errorf("unable to start batch submitter: %w", err)
+	// Batcher may be enabled later
+	if !sys.cfg.DisableBatcher {
+		if err := sys.BatchSubmitter.Start(); err != nil {
+			return nil, fmt.Errorf("unable to start batch submitter: %w", err)
+		}
 	}
 
 	return sys, nil

op-e2e/system_test.go

@@ -649,7 +649,7 @@ func TestSystemMockP2P(t *testing.T) {
 	require.Contains(t, received, receiptVerif.BlockHash)
 }
 
-// TestSystemMockP2P sets up a L1 Geth node, a rollup node, and a L2 geth node and then confirms that
+// TestSystemRPCAltSync sets up a L1 Geth node, a rollup node, and a L2 geth node and then confirms that
 // the nodes can sync L2 blocks before they are confirmed on L1.
 //
 // Test steps:
@@ -660,24 +660,28 @@ func TestSystemMockP2P(t *testing.T) {
 // 6. Wait for the RPC sync method to grab the block from the sequencer over RPC and insert it into the verifier's unsafe chain.
 // 7. Wait for the verifier to sync the unsafe chain into the safe chain.
 // 8. Verify that the TX is included in the verifier's safe chain.
-func TestSystemMockAltSync(t *testing.T) {
+func TestSystemRPCAltSync(t *testing.T) {
 	parallel(t)
 	if !verboseGethNodes {
 		log.Root().SetHandler(log.DiscardHandler())
 	}
 
 	cfg := DefaultSystemConfig(t)
-	// slow down L1 blocks so we can see the L2 blocks arrive well before the L1 blocks do.
-	// Keep the seq window small so the L2 chain is started quick
-	cfg.DeployConfig.L1BlockTime = 10
+	// the default is nil, but this may change in the future.
+	// This test must ensure the blocks are not synced via Gossip, but instead via the alt RPC based sync.
+	cfg.P2PTopology = nil
+	// Disable batcher, so there will not be any L1 data to sync from
+	cfg.DisableBatcher = true
 
-	var published, received []common.Hash
+	var published, received []string
 	seqTracer, verifTracer := new(FnTracer), new(FnTracer)
+	// The sequencer still publishes the blocks to the tracer, even if they do not reach the network due to disabled P2P
 	seqTracer.OnPublishL2PayloadFn = func(ctx context.Context, payload *eth.ExecutionPayload) {
-		published = append(published, payload.BlockHash)
+		published = append(published, payload.ID().String())
 	}
+	// Blocks are now received via the RPC based alt-sync method
 	verifTracer.OnUnsafeL2PayloadFn = func(ctx context.Context, from peer.ID, payload *eth.ExecutionPayload) {
-		received = append(received, payload.BlockHash)
+		received = append(received, payload.ID().String())
 	}
 	cfg.Nodes["sequencer"].Tracer = seqTracer
 	cfg.Nodes["verifier"].Tracer = verifTracer
@@ -687,8 +691,8 @@ func TestSystemMockAltSync(t *testing.T) {
 		role: "sequencer",
 		action: func(sCfg *SystemConfig, system *System) {
 			rpc, _ := system.Nodes["sequencer"].Attach() // never errors
-			cfg.Nodes["verifier"].L2Sync = &rollupNode.L2SyncRPCConfig{
-				Rpc: client.NewBaseRPCClient(rpc),
+			cfg.Nodes["verifier"].L2Sync = &rollupNode.PreparedL2SyncEndpoint{
+				Client: client.NewBaseRPCClient(rpc),
 			}
 		},
 	})
@@ -726,7 +730,7 @@ func TestSystemMockAltSync(t *testing.T) {
 	require.Equal(t, receiptSeq, receiptVerif)
 
 	// Verify that the tx was received via RPC sync (P2P is disabled)
-	require.Contains(t, received, receiptVerif.BlockHash)
+	require.Contains(t, received, eth.BlockID{Hash: receiptVerif.BlockHash, Number: receiptVerif.BlockNumber.Uint64()}.String())
 
 	// Verify that everything that was received was published
 	require.GreaterOrEqual(t, len(published), len(received))

op-node/eth/label.go

@@ -17,3 +17,9 @@ const (
 	// - L2: Derived chain tip from finalized L1 data
 	Finalized = "finalized"
 )
+
+func (label BlockLabel) Arg() any { return string(label) }
+
+func (BlockLabel) CheckID(id BlockID) error {
+	return nil
+}

op-node/flags/flags.go

@@ -175,6 +175,13 @@ var (
 		EnvVar:   prefixEnvVar("L2_BACKUP_UNSAFE_SYNC_RPC"),
 		Required: false,
 	}
+	BackupL2UnsafeSyncRPCTrustRPC = cli.StringFlag{
+		Name: "l2.backup-unsafe-sync-rpc.trustrpc",
+		Usage: "Like l1.trustrpc, configure if response data from the RPC needs to be verified, e.g. blockhash computation." +
+			"This does not include checks if the blockhash is part of the canonical chain.",
+		EnvVar:   prefixEnvVar("L2_BACKUP_UNSAFE_SYNC_RPC_TRUST_RPC"),
+		Required: false,
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -207,6 +214,7 @@ var optionalFlags = []cli.Flag{
 	HeartbeatMonikerFlag,
 	HeartbeatURLFlag,
 	BackupL2UnsafeSyncRPC,
+	BackupL2UnsafeSyncRPCTrustRPC,
 }
 
 // Flags contains the list of configuration options available to the binary.

op-node/flags/p2p_flags.go

@@ -34,6 +34,15 @@ var (
 		Value:    "none",
 		EnvVar:   p2pEnv("PEER_SCORING"),
 	}
+	PeerScoreBands = cli.StringFlag{
+		Name: "p2p.score.bands",
+		Usage: "Sets the peer score bands used primarily for peer score metrics. " +
+			"Should be provided in following format: <threshold>:<label>;<threshold>:<label>;..." +
+			"For example: -40:graylist;-20:restricted;0:nopx;20:friend;",
+		Required: false,
+		Value:    "-40:graylist;-20:restricted;0:nopx;20:friend;",
+		EnvVar:   p2pEnv("SCORE_BANDS"),
+	}
 
 	// Banning Flag - whether or not we want to act on the scoring
 	Banning = cli.BoolFlag{
@@ -276,6 +285,10 @@ var p2pFlags = []cli.Flag{
 	NoDiscovery,
 	P2PPrivPath,
 	P2PPrivRaw,
+	PeerScoring,
+	PeerScoreBands,
+	Banning,
+	TopicScoring,
 	ListenIP,
 	ListenTCPPort,
 	ListenUDPPort,

op-node/metrics/metrics.go

@@ -15,7 +15,6 @@ import (
 
 	pb "github.com/libp2p/go-libp2p-pubsub/pb"
 	libp2pmetrics "github.com/libp2p/go-libp2p/core/metrics"
-	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -66,7 +65,7 @@ type Metricer interface {
 	RecordSequencerSealingTime(duration time.Duration)
 	Document() []metrics.DocumentedMetric
 	// P2P Metrics
-	RecordPeerScoring(peerID peer.ID, score float64)
+	SetPeerScores(scores map[string]float64)
 }
 
 // Metrics tracks all the metrics for the op-node.
@@ -287,21 +286,24 @@ func NewMetrics(procName string) *Metrics {
 			Name:      "peer_count",
 			Help:      "Count of currently connected p2p peers",
 		}),
-		StreamCount: factory.NewGauge(prometheus.GaugeOpts{
-			Namespace: ns,
-			Subsystem: "p2p",
-			Name:      "stream_count",
-			Help:      "Count of currently connected p2p streams",
-		}),
+		// Notice: We cannot use peer ids as [Labels] in the GaugeVec
+		// since peer ids would open a service attack vector.
+		// Each peer id would be a separate metric, flooding prometheus.
+		//
+		// [Labels]: https://prometheus.io/docs/practices/naming/#labels
 		PeerScores: factory.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: ns,
 			Subsystem: "p2p",
 			Name:      "peer_scores",
-			Help:      "Peer scoring",
+			Help:      "Count of peer scores grouped by score",
 		}, []string{
-			// No label names here since peer ids would open a service attack vector.
-			// Each peer id would be a separate metric, flooding prometheus.
-			// See: https://prometheus.io/docs/practices/naming/#labels
+			"band",
+		}),
+		StreamCount: factory.NewGauge(prometheus.GaugeOpts{
+			Namespace: ns,
+			Subsystem: "p2p",
+			Name:      "stream_count",
+			Help:      "Count of currently connected p2p streams",
 		}),
 		GossipEventsTotal: factory.NewCounterVec(prometheus.CounterOpts{
 			Namespace: ns,
@@ -350,6 +352,14 @@ func NewMetrics(procName string) *Metrics {
 	}
 }
 
+// SetPeerScores updates the peer score [prometheus.GaugeVec].
+// This takes a map of labels to scores.
+func (m *Metrics) SetPeerScores(scores map[string]float64) {
+	for label, score := range scores {
+		m.PeerScores.WithLabelValues(label).Set(score)
+	}
+}
+
 // RecordInfo sets a pseudo-metric that contains versioning and
 // config info for the opnode.
 func (m *Metrics) RecordInfo(version string) {
@@ -491,10 +501,6 @@ func (m *Metrics) RecordGossipEvent(evType int32) {
 	m.GossipEventsTotal.WithLabelValues(pb.TraceEvent_Type_name[evType]).Inc()
 }
 
-func (m *Metrics) RecordPeerScoring(peerID peer.ID, score float64) {
-	m.PeerScores.WithLabelValues(peerID.String()).Set(score)
-}
-
 func (m *Metrics) IncPeerCount() {
 	m.PeerCount.Inc()
 }
@@ -627,7 +633,7 @@ func (n *noopMetricer) RecordSequencerReset() {
 func (n *noopMetricer) RecordGossipEvent(evType int32) {
 }
 
-func (n *noopMetricer) RecordPeerScoring(peerID peer.ID, score float64) {
+func (n *noopMetricer) SetPeerScores(scores map[string]float64) {
 }
 
 func (n *noopMetricer) IncPeerCount() {

op-node/node/client.go

@@ -20,7 +20,9 @@ type L2EndpointSetup interface {
 }
 
 type L2SyncEndpointSetup interface {
-	Setup(ctx context.Context, log log.Logger) (cl client.RPC, err error)
+	// Setup a RPC client to another L2 node to sync L2 blocks from.
+	// It may return a nil client with nil error if RPC based sync is not enabled.
+	Setup(ctx context.Context, log log.Logger) (cl client.RPC, trust bool, err error)
 	Check() error
 }
 
@@ -82,45 +84,45 @@ func (p *PreparedL2Endpoints) Setup(ctx context.Context, log log.Logger) (client
 
 // L2SyncEndpointConfig contains configuration for the fallback sync endpoint
 type L2SyncEndpointConfig struct {
-	// Address of the L2 RPC to use for backup sync
+	// Address of the L2 RPC to use for backup sync, may be empty if RPC alt-sync is disabled.
 	L2NodeAddr string
+	TrustRPC   bool
 }
 
 var _ L2SyncEndpointSetup = (*L2SyncEndpointConfig)(nil)
 
-func (cfg *L2SyncEndpointConfig) Setup(ctx context.Context, log log.Logger) (client.RPC, error) {
+// Setup creates an RPC client to sync from.
+// It will return nil without error if no sync method is configured.
+func (cfg *L2SyncEndpointConfig) Setup(ctx context.Context, log log.Logger) (cl client.RPC, trust bool, err error) {
+	if cfg.L2NodeAddr == "" {
+		return nil, false, nil
+	}
 	l2Node, err := client.NewRPC(ctx, log, cfg.L2NodeAddr)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
-	return l2Node, nil
+	return l2Node, cfg.TrustRPC, nil
 }
 
 func (cfg *L2SyncEndpointConfig) Check() error {
-	if cfg.L2NodeAddr == "" {
-		return errors.New("empty L2 Node Address")
-	}
-
+	// empty addr is valid, as it is optional.
 	return nil
 }
 
-type L2SyncRPCConfig struct {
-	// RPC endpoint to use for syncing
-	Rpc client.RPC
+type PreparedL2SyncEndpoint struct {
+	// RPC endpoint to use for syncing, may be nil if RPC alt-sync is disabled.
+	Client   client.RPC
+	TrustRPC bool
 }
 
-var _ L2SyncEndpointSetup = (*L2SyncRPCConfig)(nil)
+var _ L2SyncEndpointSetup = (*PreparedL2SyncEndpoint)(nil)
 
-func (cfg *L2SyncRPCConfig) Setup(ctx context.Context, log log.Logger) (client.RPC, error) {
-	return cfg.Rpc, nil
+func (cfg *PreparedL2SyncEndpoint) Setup(ctx context.Context, log log.Logger) (cl client.RPC, trust bool, err error) {
+	return cfg.Client, cfg.TrustRPC, nil
 }
 
-func (cfg *L2SyncRPCConfig) Check() error {
-	if cfg.Rpc == nil {
-		return errors.New("rpc cannot be nil")
-	}
-
+func (cfg *PreparedL2SyncEndpoint) Check() error {
 	return nil
 }
 

op-node/node/config.go

@@ -80,6 +80,9 @@ func (cfg *Config) Check() error {
 	if err := cfg.L2.Check(); err != nil {
 		return fmt.Errorf("l2 endpoint config error: %w", err)
 	}
+	if err := cfg.L2Sync.Check(); err != nil {
+		return fmt.Errorf("sync config error: %w", err)
+	}
 	if err := cfg.Rollup.Check(); err != nil {
 		return fmt.Errorf("rollup config error: %w", err)
 	}

op-node/node/node.go

@@ -33,6 +33,7 @@ type OpNode struct {
 	l1Source  *sources.L1Client     // L1 Client to fetch data from
 	l2Driver  *driver.Driver        // L2 Engine to Sync
 	l2Source  *sources.EngineClient // L2 Execution Engine RPC bindings
+	rpcSync   *sources.SyncClient   // Alt-sync RPC client, optional (may be nil)
 	server    *rpcServer            // RPC server hosting the rollup-node API
 	p2pNode   *p2p.NodeP2P          // P2P node functionality
 	p2pSigner p2p.Signer            // p2p gogssip application messages will be signed with this signer
@@ -86,6 +87,9 @@ func (n *OpNode) init(ctx context.Context, cfg *Config, snapshotLog log.Logger)
 	if err := n.initL2(ctx, cfg, snapshotLog); err != nil {
 		return err
 	}
+	if err := n.initRPCSync(ctx, cfg); err != nil {
+		return err
+	}
 	if err := n.initP2PSigner(ctx, cfg); err != nil {
 		return err
 	}
@@ -197,29 +201,27 @@ func (n *OpNode) initL2(ctx context.Context, cfg *Config, snapshotLog log.Logger
 		return err
 	}
 
-	var syncClient *sources.SyncClient
-	// If the L2 sync config is present, use it to create a sync client
-	if cfg.L2Sync != nil {
-		if err := cfg.L2Sync.Check(); err != nil {
-			log.Info("L2 sync config is not present, skipping L2 sync client setup", "err", err)
-		} else {
-			rpcSyncClient, err := cfg.L2Sync.Setup(ctx, n.log)
-			if err != nil {
-				return fmt.Errorf("failed to setup L2 execution-engine RPC client for backup sync: %w", err)
-			}
+	n.l2Driver = driver.NewDriver(&cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source, n, n, n.log, snapshotLog, n.metrics)
 
-			// The sync client's RPC is always trusted
-			config := sources.SyncClientDefaultConfig(&cfg.Rollup, true)
+	return nil
+}
 
-			syncClient, err = sources.NewSyncClient(n.OnUnsafeL2Payload, rpcSyncClient, n.log, n.metrics.L2SourceCache, config)
-			if err != nil {
-				return fmt.Errorf("failed to create sync client: %w", err)
-			}
-		}
+func (n *OpNode) initRPCSync(ctx context.Context, cfg *Config) error {
+	rpcSyncClient, trustRPC, err := cfg.L2Sync.Setup(ctx, n.log)
+	if err != nil {
+		return fmt.Errorf("failed to setup L2 execution-engine RPC client for backup sync: %w", err)
+	}
+	if rpcSyncClient == nil { // if no RPC client is configured to sync from, then don't add the RPC sync client
+		return nil
 	}
 
-	n.l2Driver = driver.NewDriver(&cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source, syncClient, n, n.log, snapshotLog, n.metrics)
+	config := sources.SyncClientDefaultConfig(&cfg.Rollup, trustRPC)
 
+	syncClient, err := sources.NewSyncClient(n.OnUnsafeL2Payload, rpcSyncClient, n.log, n.metrics.L2SourceCache, config)
+	if err != nil {
+		return fmt.Errorf("failed to create sync client: %w", err)
+	}
+	n.rpcSync = syncClient
 	return nil
 }
 
@@ -292,11 +294,12 @@ func (n *OpNode) Start(ctx context.Context) error {
 	}
 
 	// If the backup unsafe sync client is enabled, start its event loop
-	if n.l2Driver.L2SyncCl != nil {
-		if err := n.l2Driver.L2SyncCl.Start(); err != nil {
+	if n.rpcSync != nil {
+		if err := n.rpcSync.Start(); err != nil {
 			n.log.Error("Could not start the backup sync client", "err", err)
 			return err
 		}
+		n.log.Info("Started L2-RPC sync service")
 	}
 
 	return nil
@@ -375,6 +378,14 @@ func (n *OpNode) OnUnsafeL2Payload(ctx context.Context, from peer.ID, payload *e
 	return nil
 }
 
+func (n *OpNode) RequestL2Range(ctx context.Context, start, end uint64) error {
+	if n.rpcSync != nil {
+		return n.rpcSync.RequestL2Range(ctx, start, end)
+	}
+	n.log.Debug("ignoring request to sync L2 range, no sync method available")
+	return nil
+}
+
 func (n *OpNode) P2P() p2p.Node {
 	return n.p2pNode
 }
@@ -413,8 +424,8 @@ func (n *OpNode) Close() error {
 		}
 
 		// If the L2 sync client is present & running, close it.
-		if n.l2Driver.L2SyncCl != nil {
-			if err := n.l2Driver.L2SyncCl.Close(); err != nil {
+		if n.rpcSync != nil {
+			if err := n.rpcSync.Close(); err != nil {
 				result = multierror.Append(result, fmt.Errorf("failed to close L2 engine backup sync client cleanly: %w", err))
 			}
 		}

op-node/p2p/band_scorer_test.go

+package p2p
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestBandScorer_ParseDefault tests the [BandScorer.Parse] function
+// on the default band scores cli flag value.
+func TestBandScorer_ParseDefault(t *testing.T) {
+	// Create a new band scorer.
+	bandScorer, err := NewBandScorer("-40:graylist;-20:restricted;0:nopx;20:friend;")
+	require.NoError(t, err)
+
+	// Validate the [BandScorer] internals.
+	require.ElementsMatch(t, bandScorer.bands, []scorePair{
+		{band: "graylist", threshold: -40},
+		{band: "restricted", threshold: -20},
+		{band: "nopx", threshold: 0},
+		{band: "friend", threshold: 20},
+	})
+}
+
+// TestBandScorer_BucketCorrectly tests the [BandScorer.Bucket] function
+// on a variety of scores.
+func TestBandScorer_BucketCorrectly(t *testing.T) {
+	// Create a new band scorer.
+	bandScorer, err := NewBandScorer("-40:graylist;-20:restricted;0:nopx;20:friend;")
+	require.NoError(t, err)
+
+	// Validate the [BandScorer] internals.
+	require.Equal(t, bandScorer.Bucket(-100), "graylist")
+	require.Equal(t, bandScorer.Bucket(-40), "graylist")
+	require.Equal(t, bandScorer.Bucket(-39), "restricted")
+	require.Equal(t, bandScorer.Bucket(-20), "restricted")
+	require.Equal(t, bandScorer.Bucket(-19), "nopx")
+	require.Equal(t, bandScorer.Bucket(0), "nopx")
+	require.Equal(t, bandScorer.Bucket(1), "friend")
+	require.Equal(t, bandScorer.Bucket(20), "friend")
+	require.Equal(t, bandScorer.Bucket(21), "friend")
+}
+
+// TestBandScorer_BucketInverted tests the [BandScorer.Bucket] function
+// on a variety of scores, in descending order.
+func TestBandScorer_BucketInverted(t *testing.T) {
+	// Create a new band scorer.
+	bandScorer, err := NewBandScorer("20:friend;0:nopx;-20:restricted;-40:graylist;")
+	require.NoError(t, err)
+
+	// Validate the [BandScorer] internals.
+	require.Equal(t, bandScorer.Bucket(-100), "graylist")
+	require.Equal(t, bandScorer.Bucket(-40), "graylist")
+	require.Equal(t, bandScorer.Bucket(-39), "restricted")
+	require.Equal(t, bandScorer.Bucket(-20), "restricted")
+	require.Equal(t, bandScorer.Bucket(-19), "nopx")
+	require.Equal(t, bandScorer.Bucket(0), "nopx")
+	require.Equal(t, bandScorer.Bucket(1), "friend")
+	require.Equal(t, bandScorer.Bucket(20), "friend")
+	require.Equal(t, bandScorer.Bucket(21), "friend")
+}
+
+// TestBandScorer_ParseEmpty tests the [BandScorer.Parse] function
+// on an empty string.
+func TestBandScorer_ParseEmpty(t *testing.T) {
+	// Create a band scorer on an empty string.
+	bandScorer, err := NewBandScorer("")
+	require.NoError(t, err)
+
+	// Validate the [BandScorer] internals.
+	require.Len(t, bandScorer.bands, 0)
+}
+
+// TestBandScorer_ParseWhitespace tests the [BandScorer.Parse] function
+// on a variety of whitespaced strings.
+func TestBandScorer_ParseWhitespace(t *testing.T) {
+	// Create a band scorer on an empty string.
+	bandScorer, err := NewBandScorer("  ;  ;  ;  ")
+	require.NoError(t, err)
+
+	// Validate the [BandScorer] internals.
+	require.Len(t, bandScorer.bands, 0)
+}

op-node/p2p/cli/load_config.go

@@ -58,6 +58,10 @@ func NewConfig(ctx *cli.Context, blockTime uint64) (*p2p.Config, error) {
 		return nil, fmt.Errorf("failed to load p2p peer scoring options: %w", err)
 	}
 
+	if err := loadPeerScoreBands(conf, ctx); err != nil {
+		return nil, fmt.Errorf("failed to load p2p peer score bands: %w", err)
+	}
+
 	if err := loadBanningOption(conf, ctx); err != nil {
 		return nil, fmt.Errorf("failed to load banning option: %w", err)
 	}
@@ -121,6 +125,17 @@ func loadPeerScoringParams(conf *p2p.Config, ctx *cli.Context, blockTime uint64)
 	return nil
 }
 
+// loadPeerScoreBands loads [p2p.BandScorer] from the CLI context.
+func loadPeerScoreBands(conf *p2p.Config, ctx *cli.Context) error {
+	scoreBands := ctx.GlobalString(flags.PeerScoreBands.Name)
+	bandScorer, err := p2p.NewBandScorer(scoreBands)
+	if err != nil {
+		return err
+	}
+	conf.BandScoreThresholds = *bandScorer
+	return nil
+}
+
 // loadBanningOption loads whether or not to ban peers from the CLI context.
 func loadBanningOption(conf *p2p.Config, ctx *cli.Context) error {
 	ban := ctx.GlobalBool(flags.Banning.Name)

op-node/p2p/config.go

@@ -54,6 +54,9 @@ type Config struct {
 	PeerScoring  pubsub.PeerScoreParams
 	TopicScoring pubsub.TopicScoreParams
 
+	// Peer Score Band Thresholds
+	BandScoreThresholds BandScoreThresholds
+
 	// Whether to ban peers based on their [PeerScoring] score.
 	BanningEnabled bool
 
@@ -151,6 +154,10 @@ func (conf *Config) PeerScoringParams() *pubsub.PeerScoreParams {
 	return &conf.PeerScoring
 }
 
+func (conf *Config) PeerBandScorer() *BandScoreThresholds {
+	return &conf.BandScoreThresholds
+}
+
 func (conf *Config) BanPeers() bool {
 	return conf.BanningEnabled
 }

op-node/p2p/gossip.go

@@ -55,6 +55,7 @@ type GossipSetupConfigurables interface {
 	TopicScoringParams() *pubsub.TopicScoreParams
 	BanPeers() bool
 	ConfigureGossip(params *pubsub.GossipSubParams) []pubsub.Option
+	PeerBandScorer() *BandScoreThresholds
 }
 
 type GossipRuntimeConfig interface {
@@ -64,7 +65,8 @@ type GossipRuntimeConfig interface {
 //go:generate mockery --name GossipMetricer
 type GossipMetricer interface {
 	RecordGossipEvent(evType int32)
-	RecordPeerScoring(peerID peer.ID, score float64)
+	// Peer Scoring Metric Funcs
+	SetPeerScores(map[string]float64)
 }
 
 func blocksTopicV1(cfg *rollup.Config) string {

op-node/p2p/mocks/ConnectionGater.go

-// Code generated by mockery v2.14.0. DO NOT EDIT.
+// Code generated by mockery v2.22.1. DO NOT EDIT.
 
 package mocks
 
@@ -123,13 +123,16 @@ func (_m *ConnectionGater) InterceptUpgraded(_a0 network.Conn) (bool, control.Di
 	ret := _m.Called(_a0)
 
 	var r0 bool
+	var r1 control.DisconnectReason
+	if rf, ok := ret.Get(0).(func(network.Conn) (bool, control.DisconnectReason)); ok {
+		return rf(_a0)
+	}
 	if rf, ok := ret.Get(0).(func(network.Conn) bool); ok {
 		r0 = rf(_a0)
 	} else {
 		r0 = ret.Get(0).(bool)
 	}
 
-	var r1 control.DisconnectReason
 	if rf, ok := ret.Get(1).(func(network.Conn) control.DisconnectReason); ok {
 		r1 = rf(_a0)
 	} else {

op-node/p2p/mocks/GossipMetricer.go

-// Code generated by mockery v2.14.0. DO NOT EDIT.
+// Code generated by mockery v2.22.1. DO NOT EDIT.
 
 package mocks
 
-import (
-	mock "github.com/stretchr/testify/mock"
-
-	peer "github.com/libp2p/go-libp2p/core/peer"
-)
+import mock "github.com/stretchr/testify/mock"
 
 // GossipMetricer is an autogenerated mock type for the GossipMetricer type
 type GossipMetricer struct {
@@ -18,9 +14,9 @@ func (_m *GossipMetricer) RecordGossipEvent(evType int32) {
 	_m.Called(evType)
 }
 
-// RecordPeerScoring provides a mock function with given fields: peerID, score
-func (_m *GossipMetricer) RecordPeerScoring(peerID peer.ID, score float64) {
-	_m.Called(peerID, score)
+// SetPeerScores provides a mock function with given fields: _a0
+func (_m *GossipMetricer) SetPeerScores(_a0 map[string]float64) {
+	_m.Called(_a0)
 }
 
 type mockConstructorTestingTNewGossipMetricer interface {

op-node/p2p/mocks/PeerGater.go

-// Code generated by mockery v2.14.0. DO NOT EDIT.
+// Code generated by mockery v2.22.1. DO NOT EDIT.
 
 package mocks
 

op-node/p2p/mocks/Peerstore.go

-// Code generated by mockery v2.14.0. DO NOT EDIT.
+// Code generated by mockery v2.22.1. DO NOT EDIT.
 
 package mocks
 

op-node/p2p/peer_gater.go

@@ -43,11 +43,15 @@ func (s *gater) Update(id peer.ID, score float64) {
 	if score < PeerScoreThreshold && s.banEnabled {
 		s.log.Warn("peer blocking enabled, blocking peer", "id", id.String(), "score", score)
 		err := s.connGater.BlockPeer(id)
-		s.log.Warn("connection gater failed to block peer", id.String(), "err", err)
+		if err != nil {
+			s.log.Warn("connection gater failed to block peer", "id", id.String(), "err", err)
+		}
 	}
 	// Unblock peers whose score has recovered to an acceptable level
 	if (score > PeerScoreThreshold) && slices.Contains(s.connGater.ListBlockedPeers(), id) {
 		err := s.connGater.UnblockPeer(id)
-		s.log.Warn("connection gater failed to unblock peer", id.String(), "err", err)
+		if err != nil {
+			s.log.Warn("connection gater failed to unblock peer", "id", id.String(), "err", err)
+		}
 	}
 }

op-node/p2p/peer_scorer.go

 package p2p
 
 import (
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	peer "github.com/libp2p/go-libp2p/core/peer"
 )
 
 type scorer struct {
-	peerStore Peerstore
-	metricer  GossipMetricer
-	log       log.Logger
-	gater     PeerGater
+	peerStore           Peerstore
+	metricer            GossipMetricer
+	log                 log.Logger
+	gater               PeerGater
+	bandScoreThresholds *BandScoreThresholds
+}
+
+// scorePair holds a band and its corresponding threshold.
+type scorePair struct {
+	band      string
+	threshold float64
+}
+
+// BandScoreThresholds holds the thresholds for classifying peers
+// into different score bands.
+type BandScoreThresholds struct {
+	bands []scorePair
+}
+
+// NewBandScorer constructs a new [BandScoreThresholds] instance.
+func NewBandScorer(str string) (*BandScoreThresholds, error) {
+	s := &BandScoreThresholds{
+		bands: make([]scorePair, 0),
+	}
+
+	for _, band := range strings.Split(str, ";") {
+		// Skip empty band strings.
+		band := strings.TrimSpace(band)
+		if band == "" {
+			continue
+		}
+		split := strings.Split(band, ":")
+		if len(split) != 2 {
+			return nil, fmt.Errorf("invalid score band: %s", band)
+		}
+		threshold, err := strconv.ParseFloat(split[0], 64)
+		if err != nil {
+			return nil, err
+		}
+		s.bands = append(s.bands, scorePair{
+			band:      split[1],
+			threshold: threshold,
+		})
+	}
+
+	// Order the bands by threshold in ascending order.
+	sort.Slice(s.bands, func(i, j int) bool {
+		return s.bands[i].threshold < s.bands[j].threshold
+	})
+
+	return s, nil
+}
+
+// Bucket returns the appropriate band for a given score.
+func (s *BandScoreThresholds) Bucket(score float64) string {
+	for _, pair := range s.bands {
+		if score <= pair.threshold {
+			return pair.band
+		}
+	}
+	// If there is no band threshold higher than the score,
+	// the peer must be placed in the highest bucket.
+	if len(s.bands) > 0 {
+		return s.bands[len(s.bands)-1].band
+	}
+	return ""
 }
 
 // Peerstore is a subset of the libp2p peerstore.Peerstore interface.
@@ -34,12 +101,13 @@ type Scorer interface {
 }
 
 // NewScorer returns a new peer scorer.
-func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer, log log.Logger) Scorer {
+func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer, bandScoreThresholds *BandScoreThresholds, log log.Logger) Scorer {
 	return &scorer{
-		peerStore: peerStore,
-		metricer:  metricer,
-		log:       log,
-		gater:     peerGater,
+		peerStore:           peerStore,
+		metricer:            metricer,
+		log:                 log,
+		gater:               peerGater,
+		bandScoreThresholds: bandScoreThresholds,
 	}
 }
 
@@ -48,13 +116,13 @@ func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer
 // The returned [pubsub.ExtendedPeerScoreInspectFn] is called with a mapping of peer IDs to peer score snapshots.
 func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
 	return func(m map[peer.ID]*pubsub.PeerScoreSnapshot) {
+		scoreMap := make(map[string]float64)
 		for id, snap := range m {
-			// Record peer score in the metricer
-			s.metricer.RecordPeerScoring(id, snap.Score)
-
-			// Update with the peer gater
+			band := s.bandScoreThresholds.Bucket(snap.Score)
+			scoreMap[band] += 1
 			s.gater.Update(id, snap.Score)
 		}
+		s.metricer.SetPeerScores(scoreMap)
 	}
 }
 

op-node/p2p/peer_scorer_test.go

@@ -20,15 +20,18 @@ type PeerScorerTestSuite struct {
 	mockGater    *p2pMocks.PeerGater
 	mockStore    *p2pMocks.Peerstore
 	mockMetricer *p2pMocks.GossipMetricer
+	bandScorer   *p2p.BandScoreThresholds
 	logger       log.Logger
 }
 
 // SetupTest sets up the test suite.
 func (testSuite *PeerScorerTestSuite) SetupTest() {
 	testSuite.mockGater = &p2pMocks.PeerGater{}
-	// testSuite.mockConnGater = &p2pMocks.ConnectionGater{}
 	testSuite.mockStore = &p2pMocks.Peerstore{}
 	testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
+	bandScorer, err := p2p.NewBandScorer("0:graylist;")
+	testSuite.NoError(err)
+	testSuite.bandScorer = bandScorer
 	testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError)
 }
 
@@ -37,45 +40,49 @@ func TestPeerScorer(t *testing.T) {
 	suite.Run(t, new(PeerScorerTestSuite))
 }
 
-// TestPeerScorerOnConnect ensures we can call the OnConnect method on the peer scorer.
-func (testSuite *PeerScorerTestSuite) TestPeerScorerOnConnect() {
+// TestScorer_OnConnect ensures we can call the OnConnect method on the peer scorer.
+func (testSuite *PeerScorerTestSuite) TestScorer_OnConnect() {
 	scorer := p2p.NewScorer(
 		testSuite.mockGater,
 		testSuite.mockStore,
 		testSuite.mockMetricer,
+		testSuite.bandScorer,
 		testSuite.logger,
 	)
 	scorer.OnConnect()
 }
 
-// TestPeerScorerOnDisconnect ensures we can call the OnDisconnect method on the peer scorer.
-func (testSuite *PeerScorerTestSuite) TestPeerScorerOnDisconnect() {
+// TestScorer_OnDisconnect ensures we can call the OnDisconnect method on the peer scorer.
+func (testSuite *PeerScorerTestSuite) TestScorer_OnDisconnect() {
 	scorer := p2p.NewScorer(
 		testSuite.mockGater,
 		testSuite.mockStore,
 		testSuite.mockMetricer,
+		testSuite.bandScorer,
 		testSuite.logger,
 	)
 	scorer.OnDisconnect()
 }
 
-// TestSnapshotHook tests running the snapshot hook on the peer scorer.
-func (testSuite *PeerScorerTestSuite) TestSnapshotHook() {
+// TestScorer_SnapshotHook tests running the snapshot hook on the peer scorer.
+func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	scorer := p2p.NewScorer(
 		testSuite.mockGater,
 		testSuite.mockStore,
 		testSuite.mockMetricer,
+		testSuite.bandScorer,
 		testSuite.logger,
 	)
 	inspectFn := scorer.SnapshotHook()
 
-	// Mock the snapshot updates
-	// This doesn't return anything
-	testSuite.mockMetricer.On("RecordPeerScoring", peer.ID("peer1"), float64(-100)).Return(nil)
-
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-100)).Return(nil)
 
+	// The metricer should then be called with the peer score band map
+	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
+		"graylist": 1,
+	}).Return(nil)
+
 	// Apply the snapshot
 	snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
 		peer.ID("peer1"): {
@@ -85,24 +92,26 @@ func (testSuite *PeerScorerTestSuite) TestSnapshotHook() {
 	inspectFn(snapshotMap)
 }
 
-// TestSnapshotHookBlockPeer tests running the snapshot hook on the peer scorer with a peer score below the threshold.
+// TestScorer_SnapshotHookBlocksPeer tests running the snapshot hook on the peer scorer with a peer score below the threshold.
 // This implies that the peer should be blocked.
-func (testSuite *PeerScorerTestSuite) TestSnapshotHookBlockPeer() {
+func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHookBlocksPeer() {
 	scorer := p2p.NewScorer(
 		testSuite.mockGater,
 		testSuite.mockStore,
 		testSuite.mockMetricer,
+		testSuite.bandScorer,
 		testSuite.logger,
 	)
 	inspectFn := scorer.SnapshotHook()
 
-	// Mock the snapshot updates
-	// This doesn't return anything
-	testSuite.mockMetricer.On("RecordPeerScoring", peer.ID("peer1"), float64(-101)).Return(nil)
-
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-101)).Return(nil)
 
+	// The metricer should then be called with the peer score band map
+	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
+		"graylist": 1,
+	}).Return(nil)
+
 	// Apply the snapshot
 	snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
 		peer.ID("peer1"): {

op-node/p2p/peer_scores.go

@@ -14,7 +14,7 @@ func ConfigurePeerScoring(h host.Host, g ConnectionGater, gossipConf GossipSetup
 	peerScoreThresholds := NewPeerScoreThresholds()
 	banEnabled := gossipConf.BanPeers()
 	peerGater := NewPeerGater(g, log, banEnabled)
-	scorer := NewScorer(peerGater, h.Peerstore(), m, log)
+	scorer := NewScorer(peerGater, h.Peerstore(), m, gossipConf.PeerBandScorer(), log)
 	opts := []pubsub.Option{}
 	// Check the app specific score since libp2p doesn't export it's [validate] function :/
 	if peerScoreParams != nil && peerScoreParams.AppSpecificScore != nil {

op-node/p2p/peer_scores_test.go

@@ -11,7 +11,7 @@ import (
 	p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
 	testlog "github.com/ethereum-optimism/optimism/op-node/testlog"
 
-	mock "github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/mock"
 	suite "github.com/stretchr/testify/suite"
 
 	log "github.com/ethereum/go-ethereum/log"
@@ -30,6 +30,7 @@ type PeerScoresTestSuite struct {
 	mockGater    *p2pMocks.ConnectionGater
 	mockStore    *p2pMocks.Peerstore
 	mockMetricer *p2pMocks.GossipMetricer
+	bandScorer   p2p.BandScoreThresholds
 	logger       log.Logger
 }
 
@@ -38,6 +39,9 @@ func (testSuite *PeerScoresTestSuite) SetupTest() {
 	testSuite.mockGater = &p2pMocks.ConnectionGater{}
 	testSuite.mockStore = &p2pMocks.Peerstore{}
 	testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
+	bandScorer, err := p2p.NewBandScorer("0:graylist;")
+	testSuite.NoError(err)
+	testSuite.bandScorer = *bandScorer
 	testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError)
 }
 
@@ -68,6 +72,7 @@ func newGossipSubs(testSuite *PeerScoresTestSuite, ctx context.Context, hosts []
 		rt := pubsub.DefaultGossipSubRouter(h)
 		opts := []pubsub.Option{}
 		opts = append(opts, p2p.ConfigurePeerScoring(h, testSuite.mockGater, &p2p.Config{
+			BandScoreThresholds: testSuite.bandScorer,
 			PeerScoring: pubsub.PeerScoreParams{
 				AppSpecificScore: func(p peer.ID) float64 {
 					if p == hosts[0].ID() {
@@ -118,8 +123,7 @@ func (testSuite *PeerScoresTestSuite) TestNegativeScores() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	testSuite.mockMetricer.On("RecordPeerScoring", mock.Anything, float64(0)).Return(nil)
-	testSuite.mockMetricer.On("RecordPeerScoring", mock.Anything, float64(-1000)).Return(nil)
+	testSuite.mockMetricer.On("SetPeerScores", mock.Anything).Return(nil)
 
 	testSuite.mockGater.On("ListBlockedPeers").Return([]peer.ID{})
 

op-node/p2p/prepared.go

@@ -68,6 +68,10 @@ func (p *Prepared) PeerScoringParams() *pubsub.PeerScoreParams {
 	return nil
 }
 
+func (p *Prepared) PeerBandScorer() *BandScoreThresholds {
+	return nil
+}
+
 func (p *Prepared) BanPeers() bool {
 	return false
 }

op-node/rollup/derive/engine_queue.go

@@ -107,7 +107,7 @@ type EngineQueue struct {
 	// The queued-up attributes
 	safeAttributesParent eth.L2BlockRef
 	safeAttributes       *eth.PayloadAttributes
-	unsafePayloads       PayloadsQueue // queue of unsafe payloads, ordered by ascending block number, may have gaps
+	unsafePayloads       *PayloadsQueue // queue of unsafe payloads, ordered by ascending block number, may have gaps and duplicates
 
 	// Tracks which L2 blocks where last derived from which L1 block. At most finalityLookback large.
 	finalityData []FinalityData
@@ -127,18 +127,14 @@ var _ EngineControl = (*EngineQueue)(nil)
 // NewEngineQueue creates a new EngineQueue, which should be Reset(origin) before use.
 func NewEngineQueue(log log.Logger, cfg *rollup.Config, engine Engine, metrics Metrics, prev NextAttributesProvider, l1Fetcher L1Fetcher) *EngineQueue {
 	return &EngineQueue{
-		log:          log,
-		cfg:          cfg,
-		engine:       engine,
-		metrics:      metrics,
-		finalityData: make([]FinalityData, 0, finalityLookback),
-		unsafePayloads: PayloadsQueue{
-			MaxSize:  maxUnsafePayloadsMemory,
-			SizeFn:   payloadMemSize,
-			blockNos: make(map[uint64]bool),
-		},
-		prev:      prev,
-		l1Fetcher: l1Fetcher,
+		log:            log,
+		cfg:            cfg,
+		engine:         engine,
+		metrics:        metrics,
+		finalityData:   make([]FinalityData, 0, finalityLookback),
+		unsafePayloads: NewPayloadsQueue(maxUnsafePayloadsMemory, payloadMemSize),
+		prev:           prev,
+		l1Fetcher:      l1Fetcher,
 	}
 }
 
@@ -682,7 +678,8 @@ func (eq *EngineQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.System
 	return io.EOF
 }
 
-// GetUnsafeQueueGap retrieves the current [start, end] range of the gap between the tip of the unsafe priority queue and the unsafe head.
+// GetUnsafeQueueGap retrieves the current [start, end) range (incl. start, excl. end)
+// of the gap between the tip of the unsafe priority queue and the unsafe head.
 // If there is no gap, the difference between end and start will be 0.
 func (eq *EngineQueue) GetUnsafeQueueGap(expectedNumber uint64) (start uint64, end uint64) {
 	// The start of the gap is always the unsafe head + 1
@@ -691,9 +688,11 @@ func (eq *EngineQueue) GetUnsafeQueueGap(expectedNumber uint64) (start uint64, e
 	// If the priority queue is empty, the end is the first block number at the top of the priority queue
 	// Otherwise, the end is the expected block number
 	if first := eq.unsafePayloads.Peek(); first != nil {
+		// Don't include the payload we already have in the sync range
 		end = first.ID().Number
 	} else {
-		end = expectedNumber
+		// Include the expected payload in the sync range
+		end = expectedNumber + 1
 	}
 
 	return start, end

op-node/rollup/derive/payloads_queue.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/ethereum/go-ethereum/common"
+
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 )
 
@@ -48,8 +50,8 @@ func (pq *payloadsByNumber) Pop() any {
 }
 
 const (
-	// ~580 bytes per payload, with some margin for overhead
-	payloadMemFixedCost uint64 = 600
+	// ~580 bytes per payload, with some margin for overhead like map data
+	payloadMemFixedCost uint64 = 800
 	// 24 bytes per tx overhead (size of slice header in memory)
 	payloadTxMemOverhead uint64 = 24
 )
@@ -72,15 +74,25 @@ func payloadMemSize(p *eth.ExecutionPayload) uint64 {
 // without the need to use heap.Push/heap.Pop as caller.
 // PayloadsQueue maintains a MaxSize by counting and tracking sizes of added eth.ExecutionPayload entries.
 // When the size grows too large, the first (lowest block-number) payload is removed from the queue.
-// PayloadsQueue allows entries with same block number, or even full duplicates.
+// PayloadsQueue allows entries with same block number, but does not allow duplicate blocks
 type PayloadsQueue struct {
 	pq          payloadsByNumber
 	currentSize uint64
 	MaxSize     uint64
-	blockNos    map[uint64]bool
+	blockHashes map[common.Hash]struct{}
 	SizeFn      func(p *eth.ExecutionPayload) uint64
 }
 
+func NewPayloadsQueue(maxSize uint64, sizeFn func(p *eth.ExecutionPayload) uint64) *PayloadsQueue {
+	return &PayloadsQueue{
+		pq:          nil,
+		currentSize: 0,
+		MaxSize:     maxSize,
+		blockHashes: make(map[common.Hash]struct{}),
+		SizeFn:      sizeFn,
+	}
+}
+
 func (upq *PayloadsQueue) Len() int {
 	return len(upq.pq)
 }
@@ -100,8 +112,8 @@ func (upq *PayloadsQueue) Push(p *eth.ExecutionPayload) error {
 	if p == nil {
 		return errors.New("cannot add nil payload")
 	}
-	if upq.blockNos[p.ID().Number] {
-		return errors.New("cannot add duplicate payload")
+	if _, ok := upq.blockHashes[p.BlockHash]; ok {
+		return fmt.Errorf("cannot add duplicate payload %s", p.ID())
 	}
 	size := upq.SizeFn(p)
 	if size > upq.MaxSize {
@@ -115,7 +127,7 @@ func (upq *PayloadsQueue) Push(p *eth.ExecutionPayload) error {
 	for upq.currentSize > upq.MaxSize {
 		upq.Pop()
 	}
-	upq.blockNos[p.ID().Number] = true
+	upq.blockHashes[p.BlockHash] = struct{}{}
 	return nil
 }
 
@@ -137,7 +149,7 @@ func (upq *PayloadsQueue) Pop() *eth.ExecutionPayload {
 	}
 	ps := heap.Pop(&upq.pq).(payloadAndSize) // nosemgrep
 	upq.currentSize -= ps.size
-	// remove the key from the blockNos map
-	delete(upq.blockNos, ps.payload.ID().Number)
+	// remove the key from the block hashes map
+	delete(upq.blockHashes, ps.payload.BlockHash)
 	return ps.payload
 }

op-node/rollup/derive/payloads_queue_test.go

@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum-optimism/optimism/op-node/eth"
@@ -74,20 +75,17 @@ func TestPayloadMemSize(t *testing.T) {
 }
 
 func TestPayloadsQueue(t *testing.T) {
-	pq := PayloadsQueue{
-		MaxSize:  payloadMemFixedCost * 3,
-		SizeFn:   payloadMemSize,
-		blockNos: make(map[uint64]bool),
-	}
+	pq := NewPayloadsQueue(payloadMemFixedCost*3, payloadMemSize)
 	require.Equal(t, 0, pq.Len())
 	require.Equal(t, (*eth.ExecutionPayload)(nil), pq.Peek())
 	require.Equal(t, (*eth.ExecutionPayload)(nil), pq.Pop())
 
-	a := &eth.ExecutionPayload{BlockNumber: 3}
-	b := &eth.ExecutionPayload{BlockNumber: 4}
-	c := &eth.ExecutionPayload{BlockNumber: 5}
-	d := &eth.ExecutionPayload{BlockNumber: 6}
-	bAlt := &eth.ExecutionPayload{BlockNumber: 4}
+	a := &eth.ExecutionPayload{BlockNumber: 3, BlockHash: common.Hash{3}}
+	b := &eth.ExecutionPayload{BlockNumber: 4, BlockHash: common.Hash{4}}
+	c := &eth.ExecutionPayload{BlockNumber: 5, BlockHash: common.Hash{5}}
+	d := &eth.ExecutionPayload{BlockNumber: 6, BlockHash: common.Hash{6}}
+	bAlt := &eth.ExecutionPayload{BlockNumber: 4, BlockHash: common.Hash{0xff}}
+	bDup := &eth.ExecutionPayload{BlockNumber: 4, BlockHash: common.Hash{4}}
 	require.NoError(t, pq.Push(b))
 	require.Equal(t, pq.Len(), 1)
 	require.Equal(t, pq.Peek(), b)
@@ -130,7 +128,9 @@ func TestPayloadsQueue(t *testing.T) {
 	require.Equal(t, pq.Peek(), a)
 
 	// No duplicates allowed
-	require.Error(t, pq.Push(bAlt))
+	require.Error(t, pq.Push(bDup))
+	// But reorg data allowed
+	require.NoError(t, pq.Push(bAlt))
 
 	require.NoError(t, pq.Push(d))
 	require.Equal(t, pq.Len(), 3)

op-node/rollup/driver/driver.go

@@ -10,7 +10,6 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
-	"github.com/ethereum-optimism/optimism/op-node/sources"
 )
 
 type Metrics interface {
@@ -82,8 +81,19 @@ type Network interface {
 	PublishL2Payload(ctx context.Context, payload *eth.ExecutionPayload) error
 }
 
+type AltSync interface {
+	// RequestL2Range informs the sync source that the given range of L2 blocks is missing,
+	// and should be retrieved from any available alternative syncing source.
+	// The start of the range is inclusive, the end is exclusive.
+	// The sync results should be returned back to the driver via the OnUnsafeL2Payload(ctx, payload) method.
+	// The latest requested range should always take priority over previous requests.
+	// There may be overlaps in requested ranges.
+	// An error may be returned if the scheduling fails immediately, e.g. a context timeout.
+	RequestL2Range(ctx context.Context, start, end uint64) error
+}
+
 // NewDriver composes an events handler that tracks L1 state, triggers L2 derivation, and optionally sequences new L2 blocks.
-func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, syncClient *sources.SyncClient, network Network, log log.Logger, snapshotLog log.Logger, metrics Metrics) *Driver {
+func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, altSync AltSync, network Network, log log.Logger, snapshotLog log.Logger, metrics Metrics) *Driver {
 	l1State := NewL1State(log, metrics)
 	sequencerConfDepth := NewConfDepth(driverCfg.SequencerConfDepth, l1State.L1Head, l1)
 	findL1Origin := NewL1OriginSelector(log, cfg, sequencerConfDepth)
@@ -115,6 +125,6 @@ func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, sy
 		l1SafeSig:        make(chan eth.L1BlockRef, 10),
 		l1FinalizedSig:   make(chan eth.L1BlockRef, 10),
 		unsafeL2Payloads: make(chan *eth.ExecutionPayload, 10),
-		L2SyncCl:         syncClient,
+		altSync:          altSync,
 	}
 }

op-node/rollup/driver/state.go

@@ -16,7 +16,6 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
-	"github.com/ethereum-optimism/optimism/op-node/sources"
 	"github.com/ethereum-optimism/optimism/op-service/backoff"
 )
 
@@ -64,8 +63,8 @@ type Driver struct {
 	l1SafeSig      chan eth.L1BlockRef
 	l1FinalizedSig chan eth.L1BlockRef
 
-	// Backup unsafe sync client
-	L2SyncCl *sources.SyncClient
+	// Interface to signal the L2 block range to sync.
+	altSync AltSync
 
 	// L2 Signals:
 
@@ -200,11 +199,12 @@ func (s *Driver) eventLoop() {
 		sequencerTimer.Reset(delay)
 	}
 
-	// Create a ticker to check if there is a gap in the engine queue every 15 seconds
-	// If there is, we send requests to the backup RPC to retrieve the missing payloads
-	// and add them to the unsafe queue.
-	altSyncTicker := time.NewTicker(15 * time.Second)
+	// Create a ticker to check if there is a gap in the engine queue. Whenever
+	// there is, we send requests to sync source to retrieve the missing payloads.
+	syncCheckInterval := time.Duration(s.config.BlockTime) * time.Second * 2
+	altSyncTicker := time.NewTicker(syncCheckInterval)
 	defer altSyncTicker.Stop()
+	lastUnsafeL2 := s.derivation.UnsafeL2Head()
 
 	for {
 		// If we are sequencing, and the L1 state is ready, update the trigger for the next sequencer action.
@@ -220,6 +220,13 @@ func (s *Driver) eventLoop() {
 			sequencerCh = nil
 		}
 
+		// If the engine is not ready, or if the L2 head is actively changing, then reset the alt-sync:
+		// there is no need to request L2 blocks when we are syncing already.
+		if head := s.derivation.UnsafeL2Head(); head != lastUnsafeL2 || !s.derivation.EngineReady() {
+			lastUnsafeL2 = head
+			altSyncTicker.Reset(syncCheckInterval)
+		}
+
 		select {
 		case <-sequencerCh:
 			payload, err := s.sequencer.RunNextSequencerAction(ctx)
@@ -237,10 +244,12 @@ func (s *Driver) eventLoop() {
 			}
 			planSequencerAction() // schedule the next sequencer action to keep the sequencing looping
 		case <-altSyncTicker.C:
-			// Check if there is a gap in the current unsafe payload queue. If there is, attempt to fetch
-			// missing payloads from the backup RPC (if it is configured).
-			if s.L2SyncCl != nil {
-				s.checkForGapInUnsafeQueue(ctx)
+			// Check if there is a gap in the current unsafe payload queue.
+			ctx, cancel := context.WithTimeout(ctx, time.Second*2)
+			err := s.checkForGapInUnsafeQueue(ctx)
+			cancel()
+			if err != nil {
+				s.log.Warn("failed to check for unsafe L2 blocks to sync", "err", err)
 			}
 		case payload := <-s.unsafeL2Payloads:
 			s.snapshot("New unsafe payload")
@@ -462,35 +471,29 @@ type hashAndErrorChannel struct {
 	err  chan error
 }
 
-// checkForGapInUnsafeQueue checks if there is a gap in the unsafe queue and attempts to retrieve the missing payloads from the backup RPC.
-// WARNING: The sync client's attempt to retrieve the missing payloads is not guaranteed to succeed, and it will fail silently (besides
-// emitting warning logs) if the requests fail.
-func (s *Driver) checkForGapInUnsafeQueue(ctx context.Context) {
+// checkForGapInUnsafeQueue checks if there is a gap in the unsafe queue and attempts to retrieve the missing payloads from an alt-sync method.
+// WARNING: This is only an outgoing signal, the blocks are not guaranteed to be retrieved.
+// Results are received through OnUnsafeL2Payload.
+func (s *Driver) checkForGapInUnsafeQueue(ctx context.Context) error {
 	// subtract genesis time from wall clock to get the time elapsed since genesis, and then divide that
 	// difference by the block time to get the expected L2 block number at the current time. If the
 	// unsafe head does not have this block number, then there is a gap in the queue.
 	wallClock := uint64(time.Now().Unix())
 	genesisTimestamp := s.config.Genesis.L2Time
+	if wallClock < genesisTimestamp {
+		s.log.Debug("nothing to sync, did not reach genesis L2 time yet", "genesis", genesisTimestamp)
+		return nil
+	}
 	wallClockGenesisDiff := wallClock - genesisTimestamp
-	expectedL2Block := wallClockGenesisDiff / s.config.BlockTime
+	// Note: round down, we should not request blocks into the future.
+	blocksSinceGenesis := wallClockGenesisDiff / s.config.BlockTime
+	expectedL2Block := s.config.Genesis.L2.Number + blocksSinceGenesis
 
 	start, end := s.derivation.GetUnsafeQueueGap(expectedL2Block)
-	size := end - start
-
 	// Check if there is a gap between the unsafe head and the expected L2 block number at the current time.
-	if size > 0 {
-		s.log.Warn("Gap in payload queue tip and expected unsafe chain detected", "start", start, "end", end, "size", size)
-		s.log.Info("Attempting to fetch missing payloads from backup RPC", "start", start, "end", end, "size", size)
-
-		// Attempt to fetch the missing payloads from the backup unsafe sync RPC concurrently.
-		// Concurrent requests are safe here due to the engine queue being a priority queue.
-		for blockNumber := start; blockNumber <= end; blockNumber++ {
-			select {
-			case s.L2SyncCl.FetchUnsafeBlock <- blockNumber:
-				// Do nothing- the block number was successfully sent into the channel
-			default:
-				return // If the channel is full, return and wait for the next iteration of the event loop
-			}
-		}
+	if end > start {
+		s.log.Debug("requesting missing unsafe L2 block range", "start", start, "end", end, "size", end-start)
+		return s.altSync.RequestL2Range(ctx, start, end)
 	}
+	return nil
 }

op-node/service.go

@@ -136,6 +136,7 @@ func NewL2EndpointConfig(ctx *cli.Context, log log.Logger) (*node.L2EndpointConf
 func NewL2SyncEndpointConfig(ctx *cli.Context) *node.L2SyncEndpointConfig {
 	return &node.L2SyncEndpointConfig{
 		L2NodeAddr: ctx.GlobalString(flags.BackupL2UnsafeSyncRPC.Name),
+		TrustRPC:   ctx.GlobalBool(flags.BackupL2UnsafeSyncRPCTrustRPC.Name),
 	}
 }
 

op-node/sources/eth_client.go

@@ -165,9 +165,39 @@ func (s *EthClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Heade
 	return s.client.EthSubscribe(ctx, ch, "newHeads")
 }
 
-func (s *EthClient) headerCall(ctx context.Context, method string, id any) (*HeaderInfo, error) {
+// rpcBlockID is an internal type to enforce header and block call results match the requested identifier
+type rpcBlockID interface {
+	// Arg translates the object into an RPC argument
+	Arg() any
+	// CheckID verifies a block/header result matches the requested block identifier
+	CheckID(id eth.BlockID) error
+}
+
+// hashID implements rpcBlockID for safe block-by-hash fetching
+type hashID common.Hash
+
+func (h hashID) Arg() any { return common.Hash(h) }
+func (h hashID) CheckID(id eth.BlockID) error {
+	if common.Hash(h) != id.Hash {
+		return fmt.Errorf("expected block hash %s but got block %s", common.Hash(h), id)
+	}
+	return nil
+}
+
+// numberID implements rpcBlockID for safe block-by-number fetching
+type numberID uint64
+
+func (n numberID) Arg() any { return hexutil.EncodeUint64(uint64(n)) }
+func (n numberID) CheckID(id eth.BlockID) error {
+	if uint64(n) != id.Number {
+		return fmt.Errorf("expected block number %d but got block %s", uint64(n), id)
+	}
+	return nil
+}
+
+func (s *EthClient) headerCall(ctx context.Context, method string, id rpcBlockID) (*HeaderInfo, error) {
 	var header *rpcHeader
-	err := s.client.CallContext(ctx, &header, method, id, false) // headers are just blocks without txs
+	err := s.client.CallContext(ctx, &header, method, id.Arg(), false) // headers are just blocks without txs
 	if err != nil {
 		return nil, err
 	}
@@ -178,13 +208,16 @@ func (s *EthClient) headerCall(ctx context.Context, method string, id any) (*Hea
 	if err != nil {
 		return nil, err
 	}
+	if err := id.CheckID(eth.ToBlockID(info)); err != nil {
+		return nil, fmt.Errorf("fetched block header does not match requested ID: %w", err)
+	}
 	s.headersCache.Add(info.Hash(), info)
 	return info, nil
 }
 
-func (s *EthClient) blockCall(ctx context.Context, method string, id any) (*HeaderInfo, types.Transactions, error) {
+func (s *EthClient) blockCall(ctx context.Context, method string, id rpcBlockID) (*HeaderInfo, types.Transactions, error) {
 	var block *rpcBlock
-	err := s.client.CallContext(ctx, &block, method, id, true)
+	err := s.client.CallContext(ctx, &block, method, id.Arg(), true)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -195,14 +228,17 @@ func (s *EthClient) blockCall(ctx context.Context, method string, id any) (*Head
 	if err != nil {
 		return nil, nil, err
 	}
+	if err := id.CheckID(eth.ToBlockID(info)); err != nil {
+		return nil, nil, fmt.Errorf("fetched block data does not match requested ID: %w", err)
+	}
 	s.headersCache.Add(info.Hash(), info)
 	s.transactionsCache.Add(info.Hash(), txs)
 	return info, txs, nil
 }
 
-func (s *EthClient) payloadCall(ctx context.Context, method string, id any) (*eth.ExecutionPayload, error) {
+func (s *EthClient) payloadCall(ctx context.Context, method string, id rpcBlockID) (*eth.ExecutionPayload, error) {
 	var block *rpcBlock
-	err := s.client.CallContext(ctx, &block, method, id, true)
+	err := s.client.CallContext(ctx, &block, method, id.Arg(), true)
 	if err != nil {
 		return nil, err
 	}
@@ -213,6 +249,9 @@ func (s *EthClient) payloadCall(ctx context.Context, method string, id any) (*et
 	if err != nil {
 		return nil, err
 	}
+	if err := id.CheckID(payload.ID()); err != nil {
+		return nil, fmt.Errorf("fetched payload does not match requested ID: %w", err)
+	}
 	s.payloadsCache.Add(payload.BlockHash, payload)
 	return payload, nil
 }
@@ -231,17 +270,17 @@ func (s *EthClient) InfoByHash(ctx context.Context, hash common.Hash) (eth.Block
 	if header, ok := s.headersCache.Get(hash); ok {
 		return header.(*HeaderInfo), nil
 	}
-	return s.headerCall(ctx, "eth_getBlockByHash", hash)
+	return s.headerCall(ctx, "eth_getBlockByHash", hashID(hash))
 }
 
 func (s *EthClient) InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error) {
 	// can't hit the cache when querying by number due to reorgs.
-	return s.headerCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
+	return s.headerCall(ctx, "eth_getBlockByNumber", numberID(number))
 }
 
 func (s *EthClient) InfoByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, error) {
 	// can't hit the cache when querying the head due to reorgs / changes.
-	return s.headerCall(ctx, "eth_getBlockByNumber", string(label))
+	return s.headerCall(ctx, "eth_getBlockByNumber", label)
 }
 
 func (s *EthClient) InfoAndTxsByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, types.Transactions, error) {
@@ -250,32 +289,32 @@ func (s *EthClient) InfoAndTxsByHash(ctx context.Context, hash common.Hash) (eth
 			return header.(*HeaderInfo), txs.(types.Transactions), nil
 		}
 	}
-	return s.blockCall(ctx, "eth_getBlockByHash", hash)
+	return s.blockCall(ctx, "eth_getBlockByHash", hashID(hash))
 }
 
 func (s *EthClient) InfoAndTxsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Transactions, error) {
 	// can't hit the cache when querying by number due to reorgs.
-	return s.blockCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
+	return s.blockCall(ctx, "eth_getBlockByNumber", numberID(number))
 }
 
 func (s *EthClient) InfoAndTxsByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, types.Transactions, error) {
 	// can't hit the cache when querying the head due to reorgs / changes.
-	return s.blockCall(ctx, "eth_getBlockByNumber", string(label))
+	return s.blockCall(ctx, "eth_getBlockByNumber", label)
 }
 
 func (s *EthClient) PayloadByHash(ctx context.Context, hash common.Hash) (*eth.ExecutionPayload, error) {
 	if payload, ok := s.payloadsCache.Get(hash); ok {
 		return payload.(*eth.ExecutionPayload), nil
 	}
-	return s.payloadCall(ctx, "eth_getBlockByHash", hash)
+	return s.payloadCall(ctx, "eth_getBlockByHash", hashID(hash))
 }
 
 func (s *EthClient) PayloadByNumber(ctx context.Context, number uint64) (*eth.ExecutionPayload, error) {
-	return s.payloadCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
+	return s.payloadCall(ctx, "eth_getBlockByNumber", numberID(number))
 }
 
 func (s *EthClient) PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*eth.ExecutionPayload, error) {
-	return s.payloadCall(ctx, "eth_getBlockByNumber", string(label))
+	return s.payloadCall(ctx, "eth_getBlockByNumber", label)
 }
 
 // FetchReceipts returns a block info and all of the receipts associated with transactions in the block.

op-node/sources/eth_client_test.go

@@ -140,3 +140,40 @@ func TestEthClient_InfoByNumber(t *testing.T) {
 	require.Equal(t, info, expectedInfo)
 	m.Mock.AssertExpectations(t)
 }
+
+func TestEthClient_WrongInfoByNumber(t *testing.T) {
+	m := new(mockRPC)
+	_, rhdr := randHeader()
+	rhdr2 := *rhdr
+	rhdr2.Number += 1
+	n := rhdr.Number
+	ctx := context.Background()
+	m.On("CallContext", ctx, new(*rpcHeader),
+		"eth_getBlockByNumber", []any{n.String(), false}).Run(func(args mock.Arguments) {
+		*args[1].(**rpcHeader) = &rhdr2
+	}).Return([]error{nil})
+	s, err := NewL1Client(m, nil, nil, L1ClientDefaultConfig(&rollup.Config{SeqWindowSize: 10}, true, RPCKindBasic))
+	require.NoError(t, err)
+	_, err = s.InfoByNumber(ctx, uint64(n))
+	require.Error(t, err, "cannot accept the wrong block")
+	m.Mock.AssertExpectations(t)
+}
+
+func TestEthClient_WrongInfoByHash(t *testing.T) {
+	m := new(mockRPC)
+	_, rhdr := randHeader()
+	rhdr2 := *rhdr
+	rhdr2.Root[0] += 1
+	rhdr2.Hash = rhdr2.computeBlockHash()
+	k := rhdr.Hash
+	ctx := context.Background()
+	m.On("CallContext", ctx, new(*rpcHeader),
+		"eth_getBlockByHash", []any{k, false}).Run(func(args mock.Arguments) {
+		*args[1].(**rpcHeader) = &rhdr2
+	}).Return([]error{nil})
+	s, err := NewL1Client(m, nil, nil, L1ClientDefaultConfig(&rollup.Config{SeqWindowSize: 10}, true, RPCKindBasic))
+	require.NoError(t, err)
+	_, err = s.InfoByHash(ctx, k)
+	require.Error(t, err, "cannot accept the wrong block")
+	m.Mock.AssertExpectations(t)
+}

op-node/sources/sync_client.go

@@ -3,12 +3,17 @@ package sources
 import (
 	"context"
 	"errors"
+	"fmt"
+	"io"
 	"sync"
+	"time"
 
 	"github.com/ethereum-optimism/optimism/op-node/client"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/sources/caching"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
+
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/libp2p/go-libp2p/core/peer"
 )
@@ -18,23 +23,29 @@ var ErrNoUnsafeL2PayloadChannel = errors.New("unsafeL2Payloads channel must not
 // RpcSyncPeer is a mock PeerID for the RPC sync client.
 var RpcSyncPeer peer.ID = "ALT_RPC_SYNC"
 
+// receivePayload queues the received payload for processing.
+// This may return an error if there's no capacity for the payload.
 type receivePayload = func(ctx context.Context, from peer.ID, payload *eth.ExecutionPayload) error
 
-type SyncClientInterface interface {
+type RPCSync interface {
+	io.Closer
+	// Start starts an additional worker syncing job
 	Start() error
-	Close() error
-	fetchUnsafeBlockFromRpc(ctx context.Context, blockNumber uint64)
+	// RequestL2Range signals that the given range should be fetched, implementing the alt-sync interface.
+	RequestL2Range(ctx context.Context, start, end uint64) error
 }
 
 type SyncClient struct {
 	*L2Client
-	FetchUnsafeBlock chan uint64
-	done             chan struct{}
-	receivePayload   receivePayload
-	wg               sync.WaitGroup
-}
 
-var _ SyncClientInterface = (*SyncClient)(nil)
+	requests chan uint64
+
+	resCtx    context.Context
+	resCancel context.CancelFunc
+
+	receivePayload receivePayload
+	wg             sync.WaitGroup
+}
 
 type SyncClientConfig struct {
 	L2ClientConfig
@@ -51,41 +62,92 @@ func NewSyncClient(receiver receivePayload, client client.RPC, log log.Logger, m
 	if err != nil {
 		return nil, err
 	}
-
+	// This resource context is shared between all workers that may be started
+	resCtx, resCancel := context.WithCancel(context.Background())
 	return &SyncClient{
-		L2Client:         l2Client,
-		FetchUnsafeBlock: make(chan uint64, 128),
-		done:             make(chan struct{}),
-		receivePayload:   receiver,
+		L2Client:       l2Client,
+		resCtx:         resCtx,
+		resCancel:      resCancel,
+		requests:       make(chan uint64, 128),
+		receivePayload: receiver,
 	}, nil
 }
 
-// Start starts up the state loop.
-// The loop will have been started if err is not nil.
+// Start starts the syncing background work. This may not be called after Close().
 func (s *SyncClient) Start() error {
+	// TODO(CLI-3635): we can start multiple event loop runners as workers, to parallelize the work
 	s.wg.Add(1)
 	go s.eventLoop()
 	return nil
 }
 
-// Close sends a signal to the event loop to stop.
+// Close sends a signal to close all concurrent syncing work.
 func (s *SyncClient) Close() error {
-	s.done <- struct{}{}
+	s.resCancel()
 	s.wg.Wait()
 	return nil
 }
 
+func (s *SyncClient) RequestL2Range(ctx context.Context, start, end uint64) error {
+	// Drain previous requests now that we have new information
+	for len(s.requests) > 0 {
+		select { // in case requests is being read at the same time, don't block on draining it.
+		case <-s.requests:
+		default:
+			break
+		}
+	}
+
+	// TODO(CLI-3635): optimize the by-range fetching with the Engine API payloads-by-range method.
+
+	s.log.Info("Scheduling to fetch missing payloads from backup RPC", "start", start, "end", end, "size", end-start)
+
+	for i := start; i < end; i++ {
+		select {
+		case s.requests <- i:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return nil
+}
+
 // eventLoop is the main event loop for the sync client.
 func (s *SyncClient) eventLoop() {
 	defer s.wg.Done()
 	s.log.Info("Starting sync client event loop")
 
+	backoffStrategy := &backoff.ExponentialStrategy{
+		Min:       1000,
+		Max:       20_000,
+		MaxJitter: 250,
+	}
+
 	for {
 		select {
-		case <-s.done:
+		case <-s.resCtx.Done():
+			s.log.Debug("Shutting down RPC sync worker")
 			return
-		case blockNumber := <-s.FetchUnsafeBlock:
-			s.fetchUnsafeBlockFromRpc(context.Background(), blockNumber)
+		case reqNum := <-s.requests:
+			err := backoff.DoCtx(s.resCtx, 5, backoffStrategy, func() error {
+				// Limit the maximum time for fetching payloads
+				ctx, cancel := context.WithTimeout(s.resCtx, time.Second*10)
+				defer cancel()
+				// We are only fetching one block at a time here.
+				return s.fetchUnsafeBlockFromRpc(ctx, reqNum)
+			})
+			if err != nil {
+				if err == s.resCtx.Err() {
+					return
+				}
+				s.log.Error("failed syncing L2 block via RPC", "err", err, "num", reqNum)
+				// Reschedule at end of queue
+				select {
+				case s.requests <- reqNum:
+				default:
+					// drop syncing job if we are too busy with sync jobs already.
+				}
+			}
 		}
 	}
 }
@@ -95,28 +157,22 @@ func (s *SyncClient) eventLoop() {
 //
 // Post Shanghai hardfork, the engine API's `PayloadBodiesByRange` method will be much more efficient, but for now,
 // the `eth_getBlockByNumber` method is more widely available.
-func (s *SyncClient) fetchUnsafeBlockFromRpc(ctx context.Context, blockNumber uint64) {
+func (s *SyncClient) fetchUnsafeBlockFromRpc(ctx context.Context, blockNumber uint64) error {
 	s.log.Info("Requesting unsafe payload from backup RPC", "block number", blockNumber)
 
 	payload, err := s.PayloadByNumber(ctx, blockNumber)
 	if err != nil {
-		s.log.Warn("Failed to convert block to execution payload", "block number", blockNumber, "err", err)
-		return
-	}
-
-	// Signature validation is not necessary here since the backup RPC is trusted.
-	if _, ok := payload.CheckBlockHash(); !ok {
-		s.log.Warn("Received invalid payload from backup RPC; invalid block hash", "payload", payload.ID())
-		return
+		return fmt.Errorf("failed to fetch payload by number (%d): %w", blockNumber, err)
 	}
+	// Note: the underlying RPC client used for syncing verifies the execution payload blockhash, if set to untrusted.
 
 	s.log.Info("Received unsafe payload from backup RPC", "payload", payload.ID())
 
 	// Send the retrieved payload to the `unsafeL2Payloads` channel.
 	if err = s.receivePayload(ctx, RpcSyncPeer, payload); err != nil {
-		s.log.Warn("Failed to send payload into the driver's unsafeL2Payloads channel", "payload", payload.ID(), "err", err)
-		return
+		return fmt.Errorf("failed to send payload %s into the driver's unsafeL2Payloads channel: %w", payload.ID(), err)
 	} else {
-		s.log.Info("Sent received payload into the driver's unsafeL2Payloads channel", "payload", payload.ID())
+		s.log.Debug("Sent received payload into the driver's unsafeL2Payloads channel", "payload", payload.ID())
+		return nil
 	}
 }

op-proposer/metrics/metrics.go

+package metrics
+
+import (
+	"context"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/prometheus/client_golang/prometheus"
+
+	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
+)
+
+const Namespace = "op_proposer"
+
+type Metricer interface {
+	RecordInfo(version string)
+	RecordUp()
+
+	// Records all L1 and L2 block events
+	opmetrics.RefMetricer
+
+	RecordL2BlocksProposed(l2ref eth.L2BlockRef)
+}
+
+type Metrics struct {
+	ns       string
+	registry *prometheus.Registry
+	factory  opmetrics.Factory
+
+	opmetrics.RefMetrics
+
+	Info prometheus.GaugeVec
+	Up   prometheus.Gauge
+}
+
+var _ Metricer = (*Metrics)(nil)
+
+func NewMetrics(procName string) *Metrics {
+	if procName == "" {
+		procName = "default"
+	}
+	ns := Namespace + "_" + procName
+
+	registry := opmetrics.NewRegistry()
+	factory := opmetrics.With(registry)
+
+	return &Metrics{
+		ns:       ns,
+		registry: registry,
+		factory:  factory,
+
+		RefMetrics: opmetrics.MakeRefMetrics(ns, factory),
+
+		Info: *factory.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: ns,
+			Name:      "info",
+			Help:      "Pseudo-metric tracking version and config info",
+		}, []string{
+			"version",
+		}),
+		Up: factory.NewGauge(prometheus.GaugeOpts{
+			Namespace: ns,
+			Name:      "up",
+			Help:      "1 if the op-proposer has finished starting up",
+		}),
+	}
+}
+
+func (m *Metrics) Serve(ctx context.Context, host string, port int) error {
+	return opmetrics.ListenAndServe(ctx, m.registry, host, port)
+}
+
+func (m *Metrics) StartBalanceMetrics(ctx context.Context,
+	l log.Logger, client *ethclient.Client, account common.Address) {
+	opmetrics.LaunchBalanceMetrics(ctx, l, m.registry, m.ns, client, account)
+}
+
+// RecordInfo sets a pseudo-metric that contains versioning and
+// config info for the op-proposer.
+func (m *Metrics) RecordInfo(version string) {
+	m.Info.WithLabelValues(version).Set(1)
+}
+
+// RecordUp sets the up metric to 1.
+func (m *Metrics) RecordUp() {
+	prometheus.MustRegister()
+	m.Up.Set(1)
+}
+
+const (
+	BlockProposed = "proposed"
+)
+
+// RecordL2BlocksProposed should be called when new L2 block is proposed
+func (m *Metrics) RecordL2BlocksProposed(l2ref eth.L2BlockRef) {
+	m.RecordL2Ref(BlockProposed, l2ref)
+}

op-proposer/metrics/noop.go

+package metrics
+
+import (
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
+)
+
+type noopMetrics struct{ opmetrics.NoopRefMetrics }
+
+var NoopMetrics Metricer = new(noopMetrics)
+
+func (*noopMetrics) RecordInfo(version string) {}
+func (*noopMetrics) RecordUp()                 {}
+
+func (*noopMetrics) RecordL2BlocksProposed(l2ref eth.L2BlockRef) {}

op-proposer/proposer/l2_output_submitter.go

@@ -24,9 +24,9 @@ import (
 	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
+	"github.com/ethereum-optimism/optimism/op-proposer/metrics"
 	opcrypto "github.com/ethereum-optimism/optimism/op-service/crypto"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
-	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	oppprof "github.com/ethereum-optimism/optimism/op-service/pprof"
 	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
@@ -49,9 +49,10 @@ func Main(version string, cliCtx *cli.Context) error {
 	}
 
 	l := oplog.NewLogger(cfg.LogConfig)
+	m := metrics.NewMetrics("default")
 	l.Info("Initializing L2 Output Submitter")
 
-	l2OutputSubmitter, err := NewL2OutputSubmitterFromCLIConfig(cfg, l)
+	l2OutputSubmitter, err := NewL2OutputSubmitterFromCLIConfig(cfg, l, m)
 	if err != nil {
 		l.Error("Unable to create the L2 Output Submitter", "error", err)
 		return err
@@ -78,17 +79,15 @@ func Main(version string, cliCtx *cli.Context) error {
 		}()
 	}
 
-	registry := opmetrics.NewRegistry()
 	metricsCfg := cfg.MetricsConfig
 	if metricsCfg.Enabled {
 		l.Info("starting metrics server", "addr", metricsCfg.ListenAddr, "port", metricsCfg.ListenPort)
 		go func() {
-			if err := opmetrics.ListenAndServe(ctx, registry, metricsCfg.ListenAddr, metricsCfg.ListenPort); err != nil {
+			if err := m.Serve(ctx, metricsCfg.ListenAddr, metricsCfg.ListenPort); err != nil {
 				l.Error("error starting metrics server", err)
 			}
 		}()
-		addr := l2OutputSubmitter.from
-		opmetrics.LaunchBalanceMetrics(ctx, l, registry, "", l2OutputSubmitter.l1Client, addr)
+		m.StartBalanceMetrics(ctx, l, l2OutputSubmitter.l1Client, l2OutputSubmitter.from)
 	}
 
 	rpcCfg := cfg.RPCConfig
@@ -98,6 +97,9 @@ func Main(version string, cliCtx *cli.Context) error {
 		return fmt.Errorf("error starting RPC server: %w", err)
 	}
 
+	m.RecordInfo(version)
+	m.RecordUp()
+
 	interruptChannel := make(chan os.Signal, 1)
 	signal.Notify(interruptChannel, []os.Signal{
 		os.Interrupt,
@@ -117,6 +119,7 @@ type L2OutputSubmitter struct {
 	wg    sync.WaitGroup
 	done  chan struct{}
 	log   log.Logger
+	metr  metrics.Metricer
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -143,7 +146,7 @@ type L2OutputSubmitter struct {
 }
 
 // NewL2OutputSubmitterFromCLIConfig creates a new L2 Output Submitter given the CLI Config
-func NewL2OutputSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger) (*L2OutputSubmitter, error) {
+func NewL2OutputSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metricer) (*L2OutputSubmitter, error) {
 	signer, fromAddress, err := opcrypto.SignerFactoryFromConfig(l, cfg.PrivateKey, cfg.Mnemonic, cfg.L2OutputHDPath, cfg.SignerConfig)
 	if err != nil {
 		return nil, err
@@ -185,11 +188,11 @@ func NewL2OutputSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger) (*L2OutputSu
 		SignerFnFactory:    signer,
 	}
 
-	return NewL2OutputSubmitter(proposerCfg, l)
+	return NewL2OutputSubmitter(proposerCfg, l, m)
 }
 
 // NewL2OutputSubmitter creates a new L2 Output Submitter
-func NewL2OutputSubmitter(cfg Config, l log.Logger) (*L2OutputSubmitter, error) {
+func NewL2OutputSubmitter(cfg Config, l log.Logger, m metrics.Metricer) (*L2OutputSubmitter, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	cCtx, cCancel := context.WithTimeout(ctx, defaultDialTimeout)
@@ -228,6 +231,7 @@ func NewL2OutputSubmitter(cfg Config, l log.Logger) (*L2OutputSubmitter, error)
 		log:    l,
 		ctx:    ctx,
 		cancel: cancel,
+		metr:   m,
 
 		l1Client:     cfg.L1Client,
 		rollupClient: cfg.RollupClient,
@@ -413,9 +417,9 @@ func (l *L2OutputSubmitter) loop() {
 				l.log.Error("Failed to send proposal transaction", "err", err)
 				cancel()
 				break
-			} else {
-				cancel()
 			}
+			l.metr.RecordL2BlocksProposed(output.BlockRef)
+			cancel()
 
 		case <-l.done:
 			return

ops-bedrock/docker-compose.yml

@@ -61,6 +61,8 @@ services:
       --p2p.listen.ip=0.0.0.0
       --p2p.listen.tcp=9003
       --p2p.listen.udp=9003
+      --p2p.scoring.peers=light
+      --p2p.ban.peers=true
       --snapshotlog.file=/op_log/snapshot.log
       --p2p.priv.path=/config/p2p-node-key.txt
       --metrics.enabled