op-node: Remove Unsafe Sync Target & Direct Engine Manipulation (#8856)

This fully moves all execution engine manipulation into the Engine Controller
from the Engine Queue. It does not remove the proxy methods however.

op-e2e/actions/l2_verifier.go

@@ -147,10 +147,6 @@ func (s *L2Verifier) L2Unsafe() eth.L2BlockRef {
 	return s.derivation.UnsafeL2Head()
 }
 
-func (s *L2Verifier) EngineSyncTarget() eth.L2BlockRef {
-	return s.derivation.EngineSyncTarget()
-}
-
 func (s *L2Verifier) SyncStatus() *eth.SyncStatus {
 	return &eth.SyncStatus{
 		CurrentL1:          s.derivation.Origin(),
@@ -163,7 +159,6 @@ func (s *L2Verifier) SyncStatus() *eth.SyncStatus {
 		FinalizedL2:        s.L2Finalized(),
 		PendingSafeL2:      s.L2PendingSafe(),
 		UnsafeL2SyncTarget: s.derivation.UnsafeL2SyncTarget(),
-		EngineSyncTarget:   s.EngineSyncTarget(),
 	}
 }
 

op-e2e/actions/sync_test.go

@@ -132,6 +132,7 @@ func FinalizeWhileSyncing(gt *testing.T, deltaTimeOffset *hexutil.Uint64) {
 	require.Less(t, verifierStartStatus.FinalizedL2.Number, verifier.SyncStatus().FinalizedL2.Number, "verifier finalized L2 blocks during sync")
 }
 
+// TestUnsafeSync tests that a verifier properly imports unsafe blocks via gossip.
 func TestUnsafeSync(gt *testing.T) {
 	t := NewDefaultTesting(gt)
 	dp := e2eutils.MakeDeployParams(t, defaultRollupTestParams)
@@ -155,15 +156,15 @@ func TestUnsafeSync(gt *testing.T) {
 		verifier.ActL2UnsafeGossipReceive(seqHead)(t)
 		// Handle unsafe payload
 		verifier.ActL2PipelineFull(t)
-		// Verifier must advance its unsafe head and engine sync target.
+		// Verifier must advance its unsafe head.
 		require.Equal(t, sequencer.L2Unsafe().Hash, verifier.L2Unsafe().Hash)
-		// Check engine sync target updated.
-		require.Equal(t, sequencer.L2Unsafe().Hash, sequencer.EngineSyncTarget().Hash)
-		require.Equal(t, verifier.L2Unsafe().Hash, verifier.EngineSyncTarget().Hash)
 	}
 }
 
-func TestEngineP2PSync(gt *testing.T) {
+// TestELSync tests that a verifier will have the EL import the full chain from the sequencer
+// when passed a single unsafe block. op-geth can either snap sync or full sync here.
+func TestELSync(gt *testing.T) {
+	gt.Skip("not implemented yet")
 	t := NewDefaultTesting(gt)
 	dp := e2eutils.MakeDeployParams(t, defaultRollupTestParams)
 	sd := e2eutils.Setup(t, dp, defaultAlloc)
@@ -179,8 +180,6 @@ func TestEngineP2PSync(gt *testing.T) {
 	sequencer.ActL2PipelineFull(t)
 	verifier.ActL2PipelineFull(t)
 
-	verifierUnsafeHead := verifier.L2Unsafe()
-
 	// Build a L2 block. This block will not be gossiped to verifier, so verifier can not advance chain by itself.
 	sequencer.ActL2StartBlock(t)
 	sequencer.ActL2EndBlock(t)
@@ -195,12 +194,13 @@ func TestEngineP2PSync(gt *testing.T) {
 		verifier.ActL2UnsafeGossipReceive(seqHead)(t)
 		// Handle unsafe payload
 		verifier.ActL2PipelineFull(t)
-		// Verifier must advance only engine sync target.
-		require.NotEqual(t, sequencer.L2Unsafe().Hash, verifier.L2Unsafe().Hash)
-		require.NotEqual(t, verifier.L2Unsafe().Hash, verifier.EngineSyncTarget().Hash)
-		require.Equal(t, verifier.L2Unsafe().Hash, verifierUnsafeHead.Hash)
-		require.Equal(t, sequencer.L2Unsafe().Hash, verifier.EngineSyncTarget().Hash)
+		// Verifier must advance unsafe head after unsafe gossip.
+		require.Equal(t, sequencer.L2Unsafe().Hash, verifier.L2Unsafe().Hash)
 	}
+	// Actual test flow should be as follows:
+	// 1. Build a chain on the sequencer.
+	// 2. Gossip only a single final L2 block from the sequencer to the verifier.
+	// 3. Assert that the verifier has the full chain.
 }
 
 func TestInvalidPayloadInSpanBatch(gt *testing.T) {

op-node/node/server_test.go

@@ -162,7 +162,6 @@ func randomSyncStatus(rng *rand.Rand) *eth.SyncStatus {
 		FinalizedL2:        testutils.RandomL2BlockRef(rng),
 		PendingSafeL2:      testutils.RandomL2BlockRef(rng),
 		UnsafeL2SyncTarget: testutils.RandomL2BlockRef(rng),
-		EngineSyncTarget:   testutils.RandomL2BlockRef(rng),
 	}
 }
 

op-node/rollup/derive/engine_controller.go

@@ -2,14 +2,18 @@ package derive
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 )
 
+var errNoFCUNeeded = errors.New("no FCU call was needed")
+
 var _ EngineControl = (*EngineController)(nil)
 var _ LocalEngineControl = (*EngineController)(nil)
 
@@ -20,17 +24,18 @@ type ExecEngine interface {
 }
 
 type EngineController struct {
-	engine  ExecEngine // Underlying execution engine RPC
-	log     log.Logger
-	metrics Metrics
-	genesis *rollup.Genesis
+	engine   ExecEngine // Underlying execution engine RPC
+	log      log.Logger
+	metrics  Metrics
+	genesis  *rollup.Genesis
+	syncMode sync.Mode
 
 	// Block Head State
-	syncTarget      eth.L2BlockRef
 	unsafeHead      eth.L2BlockRef
 	pendingSafeHead eth.L2BlockRef
 	safeHead        eth.L2BlockRef
 	finalizedHead   eth.L2BlockRef
+	needFCUCall     bool
 
 	// Building State
 	buildingOnto eth.L2BlockRef
@@ -39,21 +44,18 @@ type EngineController struct {
 	safeAttrs    *AttributesWithParent
 }
 
-func NewEngineController(engine ExecEngine, log log.Logger, metrics Metrics, genesis rollup.Genesis) *EngineController {
+func NewEngineController(engine ExecEngine, log log.Logger, metrics Metrics, genesis rollup.Genesis, syncMode sync.Mode) *EngineController {
 	return &EngineController{
-		engine:  engine,
-		log:     log,
-		metrics: metrics,
-		genesis: &genesis,
+		engine:   engine,
+		log:      log,
+		metrics:  metrics,
+		genesis:  &genesis,
+		syncMode: syncMode,
 	}
 }
 
 // State Getters
 
-func (e *EngineController) EngineSyncTarget() eth.L2BlockRef {
-	return e.syncTarget
-}
-
 func (e *EngineController) UnsafeL2Head() eth.L2BlockRef {
 	return e.unsafeHead
 }
@@ -75,21 +77,16 @@ func (e *EngineController) BuildingPayload() (eth.L2BlockRef, eth.PayloadID, boo
 }
 
 func (e *EngineController) IsEngineSyncing() bool {
-	return e.unsafeHead.Hash != e.syncTarget.Hash
+	return false
 }
 
 // Setters
 
-// SetEngineSyncTarget implements LocalEngineControl.
-func (e *EngineController) SetEngineSyncTarget(r eth.L2BlockRef) {
-	e.metrics.RecordL2Ref("l2_engineSyncTarget", r)
-	e.syncTarget = r
-}
-
 // SetFinalizedHead implements LocalEngineControl.
 func (e *EngineController) SetFinalizedHead(r eth.L2BlockRef) {
 	e.metrics.RecordL2Ref("l2_finalized", r)
 	e.finalizedHead = r
+	e.needFCUCall = true
 }
 
 // SetPendingSafeL2Head implements LocalEngineControl.
@@ -102,12 +99,14 @@ func (e *EngineController) SetPendingSafeL2Head(r eth.L2BlockRef) {
 func (e *EngineController) SetSafeHead(r eth.L2BlockRef) {
 	e.metrics.RecordL2Ref("l2_safe", r)
 	e.safeHead = r
+	e.needFCUCall = true
 }
 
 // SetUnsafeHead implements LocalEngineControl.
 func (e *EngineController) SetUnsafeHead(r eth.L2BlockRef) {
 	e.metrics.RecordL2Ref("l2_unsafe", r)
 	e.unsafeHead = r
+	e.needFCUCall = true
 }
 
 // Engine Methods
@@ -165,7 +164,6 @@ func (e *EngineController) ConfirmPayload(ctx context.Context) (out *eth.Executi
 	}
 
 	e.unsafeHead = ref
-	e.syncTarget = ref
 
 	e.metrics.RecordL2Ref("l2_unsafe", ref)
 	e.metrics.RecordL2Ref("l2_engineSyncTarget", ref)
@@ -208,6 +206,98 @@ func (e *EngineController) resetBuildingState() {
 
 // Misc Setters only used by the engine queue
 
+// checkNewPayloadStatus checks returned status of engine_newPayloadV1 request for next unsafe payload.
+// It returns true if the status is acceptable.
+func (e *EngineController) checkNewPayloadStatus(status eth.ExecutePayloadStatus) bool {
+	if e.syncMode == sync.ELSync {
+		// Allow SYNCING and ACCEPTED if engine EL sync is enabled
+		return status == eth.ExecutionValid || status == eth.ExecutionSyncing || status == eth.ExecutionAccepted
+	}
+	return status == eth.ExecutionValid
+}
+
+// checkForkchoiceUpdatedStatus checks returned status of engine_forkchoiceUpdatedV1 request for next unsafe payload.
+// It returns true if the status is acceptable.
+func (e *EngineController) checkForkchoiceUpdatedStatus(status eth.ExecutePayloadStatus) bool {
+	if e.syncMode == sync.ELSync {
+		// Allow SYNCING if engine P2P sync is enabled
+		return status == eth.ExecutionValid || status == eth.ExecutionSyncing
+	}
+	return status == eth.ExecutionValid
+}
+
+// TryUpdateEngine attempts to update the engine with the current forkchoice state of the rollup node,
+// this is a no-op if the nodes already agree on the forkchoice state.
+func (e *EngineController) TryUpdateEngine(ctx context.Context) error {
+	if !e.needFCUCall {
+		return errNoFCUNeeded
+	}
+	if e.IsEngineSyncing() {
+		e.log.Warn("Attempting to update forkchoice state while engine is P2P syncing")
+	}
+	fc := eth.ForkchoiceState{
+		HeadBlockHash:      e.unsafeHead.Hash,
+		SafeBlockHash:      e.safeHead.Hash,
+		FinalizedBlockHash: e.finalizedHead.Hash,
+	}
+	_, err := e.engine.ForkchoiceUpdate(ctx, &fc, nil)
+	if err != nil {
+		var inputErr eth.InputError
+		if errors.As(err, &inputErr) {
+			switch inputErr.Code {
+			case eth.InvalidForkchoiceState:
+				return NewResetError(fmt.Errorf("forkchoice update was inconsistent with engine, need reset to resolve: %w", inputErr.Unwrap()))
+			default:
+				return NewTemporaryError(fmt.Errorf("unexpected error code in forkchoice-updated response: %w", err))
+			}
+		} else {
+			return NewTemporaryError(fmt.Errorf("failed to sync forkchoice with engine: %w", err))
+		}
+	}
+	e.needFCUCall = false
+	return nil
+}
+
+func (e *EngineController) InsertUnsafePayload(ctx context.Context, payload *eth.ExecutionPayload, ref eth.L2BlockRef) error {
+	status, err := e.engine.NewPayload(ctx, payload)
+	if err != nil {
+		return NewTemporaryError(fmt.Errorf("failed to update insert payload: %w", err))
+	}
+	if !e.checkNewPayloadStatus(status.Status) {
+		return NewTemporaryError(fmt.Errorf("cannot process unsafe payload: new - %v; parent: %v; err: %w",
+			payload.ID(), payload.ParentID(), eth.NewPayloadErr(payload, status)))
+	}
+
+	// Mark the new payload as valid
+	fc := eth.ForkchoiceState{
+		HeadBlockHash:      payload.BlockHash,
+		SafeBlockHash:      e.safeHead.Hash,
+		FinalizedBlockHash: e.finalizedHead.Hash,
+	}
+	fcRes, err := e.engine.ForkchoiceUpdate(ctx, &fc, nil)
+	if err != nil {
+		var inputErr eth.InputError
+		if errors.As(err, &inputErr) {
+			switch inputErr.Code {
+			case eth.InvalidForkchoiceState:
+				return NewResetError(fmt.Errorf("pre-unsafe-block forkchoice update was inconsistent with engine, need reset to resolve: %w", inputErr.Unwrap()))
+			default:
+				return NewTemporaryError(fmt.Errorf("unexpected error code in forkchoice-updated response: %w", err))
+			}
+		} else {
+			return NewTemporaryError(fmt.Errorf("failed to update forkchoice to prepare for new unsafe payload: %w", err))
+		}
+	}
+	if !e.checkForkchoiceUpdatedStatus(fcRes.PayloadStatus.Status) {
+		return NewTemporaryError(fmt.Errorf("cannot prepare unsafe chain for new payload: new - %v; parent: %v; err: %w",
+			payload.ID(), payload.ParentID(), eth.ForkchoiceUpdateErr(fcRes.PayloadStatus)))
+	}
+	e.unsafeHead = ref
+	e.needFCUCall = false
+
+	return nil
+}
+
 // ResetBuildingState implements LocalEngineControl.
 func (e *EngineController) ResetBuildingState() {
 	e.resetBuildingState()

op-node/rollup/derive/engine_queue_test.go

@@ -491,7 +491,6 @@ func TestEngineQueue_ResetWhenUnsafeOriginNotCanonical(t *testing.T) {
 	require.Equal(t, refA1, eq.Finalized(), "A1 is recognized as finalized before we run any steps")
 
 	// First step after reset will do a fork choice update
-	require.True(t, eq.needForkchoiceUpdate)
 	eng.ExpectForkchoiceUpdate(&eth.ForkchoiceState{
 		HeadBlockHash:      eq.ec.UnsafeL2Head().Hash,
 		SafeBlockHash:      eq.ec.SafeL2Head().Hash,
@@ -822,7 +821,6 @@ func TestVerifyNewL1Origin(t *testing.T) {
 			require.Equal(t, refA1, eq.Finalized(), "A1 is recognized as finalized before we run any steps")
 
 			// First step after reset will do a fork choice update
-			require.True(t, eq.needForkchoiceUpdate)
 			eng.ExpectForkchoiceUpdate(&eth.ForkchoiceState{
 				HeadBlockHash:      eq.ec.UnsafeL2Head().Hash,
 				SafeBlockHash:      eq.ec.SafeL2Head().Hash,
@@ -1084,12 +1082,19 @@ func TestResetLoop(t *testing.T) {
 
 	eq := NewEngineQueue(logger, cfg, eng, metrics.NoopMetrics, prev, l1F, &sync.Config{})
 	eq.ec.SetUnsafeHead(refA2)
-	eq.ec.SetEngineSyncTarget(refA2)
 	eq.ec.SetSafeHead(refA1)
 	eq.ec.SetFinalizedHead(refA0)
 
 	// Queue up the safe attributes
+	// Expect a FCU after during the first step
+	preFc := &eth.ForkchoiceState{
+		HeadBlockHash:      refA2.Hash,
+		SafeBlockHash:      refA1.Hash,
+		FinalizedBlockHash: refA0.Hash,
+	}
+	eng.ExpectForkchoiceUpdate(preFc, nil, nil, nil)
 	require.Nil(t, eq.safeAttributes)
+	require.ErrorIs(t, eq.Step(context.Background()), nil)
 	require.ErrorIs(t, eq.Step(context.Background()), NotEnoughData)
 	require.NotNil(t, eq.safeAttributes)
 
@@ -1097,12 +1102,12 @@ func TestResetLoop(t *testing.T) {
 	require.ErrorIs(t, eq.Reset(context.Background(), eth.L1BlockRef{}, eth.SystemConfig{}), io.EOF)
 
 	// Expect a FCU after the reset
-	preFc := &eth.ForkchoiceState{
+	postFc := &eth.ForkchoiceState{
 		HeadBlockHash:      refA2.Hash,
 		SafeBlockHash:      refA0.Hash,
 		FinalizedBlockHash: refA0.Hash,
 	}
-	eng.ExpectForkchoiceUpdate(preFc, nil, nil, nil)
+	eng.ExpectForkchoiceUpdate(postFc, nil, nil, nil)
 	require.NoError(t, eq.Step(context.Background()), "clean forkchoice state after reset")
 
 	// Crux of the test. Should be in a valid state after the reset.
@@ -1187,8 +1192,17 @@ func TestEngineQueue_StepPopOlderUnsafe(t *testing.T) {
 
 	eq.AddUnsafePayload(payloadA1)
 
-	err := eq.Step(context.Background())
-	require.NoError(t, err)
+	// First Step calls FCU
+	preFc := &eth.ForkchoiceState{
+		HeadBlockHash:      refA2.Hash,
+		SafeBlockHash:      refA0.Hash,
+		FinalizedBlockHash: refA0.Hash,
+	}
+	eng.ExpectForkchoiceUpdate(preFc, nil, nil, nil)
+	require.NoError(t, eq.Step(context.Background()))
+
+	// Second Step pops the unsafe payload
+	require.NoError(t, eq.Step(context.Background()))
 
 	require.Nil(t, eq.unsafePayloads.Peek(), "should pop the unsafe payload because it is too old")
 	fmt.Println(eq.unsafePayloads.Peek())

op-node/rollup/derive/pipeline.go

@@ -53,7 +53,6 @@ type EngineQueueStage interface {
 	UnsafeL2Head() eth.L2BlockRef
 	SafeL2Head() eth.L2BlockRef
 	PendingSafeL2Head() eth.L2BlockRef
-	EngineSyncTarget() eth.L2BlockRef
 	Origin() eth.L1BlockRef
 	SystemConfig() eth.SystemConfig
 
@@ -158,10 +157,6 @@ func (dp *DerivationPipeline) UnsafeL2Head() eth.L2BlockRef {
 	return dp.eng.UnsafeL2Head()
 }
 
-func (dp *DerivationPipeline) EngineSyncTarget() eth.L2BlockRef {
-	return dp.eng.EngineSyncTarget()
-}
-
 func (dp *DerivationPipeline) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *AttributesWithParent, updateSafe bool) (errType BlockInsertionErrType, err error) {
 	return dp.eng.StartPayload(ctx, parent, attrs, updateSafe)
 }

op-node/rollup/driver/driver.go

@@ -65,7 +65,6 @@ type DerivationPipeline interface {
 	PendingSafeL2Head() eth.L2BlockRef
 	Origin() eth.L1BlockRef
 	EngineReady() bool
-	EngineSyncTarget() eth.L2BlockRef
 }
 
 type L1StateIface interface {

op-node/rollup/driver/state.go

@@ -325,7 +325,7 @@ func (s *Driver) eventLoop() {
 				s.metrics.SetDerivationIdle(true)
 				continue
 			} else if err != nil && errors.Is(err, derive.EngineELSyncing) {
-				s.log.Debug("Derivation process went idle because the engine is syncing", "progress", s.derivation.Origin(), "sync_target", s.derivation.EngineSyncTarget(), "err", err)
+				s.log.Debug("Derivation process went idle because the engine is syncing", "progress", s.derivation.Origin(), "unsafe_head", s.derivation.UnsafeL2Head(), "err", err)
 				stepAttempts = 0
 				s.metrics.SetDerivationIdle(true)
 				continue
@@ -489,7 +489,6 @@ func (s *Driver) syncStatus() *eth.SyncStatus {
 		FinalizedL2:        s.derivation.Finalized(),
 		PendingSafeL2:      s.derivation.PendingSafeL2Head(),
 		UnsafeL2SyncTarget: s.derivation.UnsafeL2SyncTarget(),
-		EngineSyncTarget:   s.derivation.EngineSyncTarget(),
 	}
 }
 

op-service/eth/sync_status.go

@@ -37,7 +37,4 @@ type SyncStatus struct {
 	// UnsafeL2SyncTarget points to the first unprocessed unsafe L2 block.
 	// It may be zeroed if there is no targeted block.
 	UnsafeL2SyncTarget L2BlockRef `json:"queued_unsafe_l2"`
-	// EngineSyncTarget points to the L2 block that the execution engine is syncing to.
-	// If it is ahead from UnsafeL2, the engine is in progress of P2P sync.
-	EngineSyncTarget L2BlockRef `json:"engine_sync_target"`
 }

op-service/testutils/random.go

@@ -329,7 +329,6 @@ func RandomOutputResponse(rng *rand.Rand) *eth.OutputResponse {
 			SafeL2:             RandomL2BlockRef(rng),
 			FinalizedL2:        RandomL2BlockRef(rng),
 			PendingSafeL2:      RandomL2BlockRef(rng),
-			EngineSyncTarget:   RandomL2BlockRef(rng),
 		},
 	}
 }