op-node: generalize and improve alt-sync

op-e2e/migration_test.go

@@ -278,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,

op-e2e/setup.go

@@ -194,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 {
@@ -418,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
@@ -607,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/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/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/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