Move Engine State from EngineQueue to EngineController (#8824)

This moves all the control of the execution engine from the EngineQueue struct
to the EngineController struct. The Engine Controller remains in the derive
package for now to minimize the amount of changes in this PR. The EngineQueue
continues to implement the EngineControl interface & forwards all methods to
the EngineController which contains the actual implementation.

op-node/rollup/derive/engine_controller.go

+package derive
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+var _ EngineControl = (*EngineController)(nil)
+var _ LocalEngineControl = (*EngineController)(nil)
+
+type ExecEngine interface {
+	GetPayload(ctx context.Context, payloadId eth.PayloadID) (*eth.ExecutionPayload, error)
+	ForkchoiceUpdate(ctx context.Context, state *eth.ForkchoiceState, attr *eth.PayloadAttributes) (*eth.ForkchoiceUpdatedResult, error)
+	NewPayload(ctx context.Context, payload *eth.ExecutionPayload) (*eth.PayloadStatusV1, error)
+}
+
+type EngineController struct {
+	engine  ExecEngine // Underlying execution engine RPC
+	log     log.Logger
+	metrics Metrics
+	genesis *rollup.Genesis
+
+	// Block Head State
+	syncTarget      eth.L2BlockRef
+	unsafeHead      eth.L2BlockRef
+	pendingSafeHead eth.L2BlockRef
+	safeHead        eth.L2BlockRef
+	finalizedHead   eth.L2BlockRef
+
+	// Building State
+	buildingOnto eth.L2BlockRef
+	buildingID   eth.PayloadID
+	buildingSafe bool
+	safeAttrs    *AttributesWithParent
+}
+
+func NewEngineController(engine ExecEngine, log log.Logger, metrics Metrics, genesis rollup.Genesis) *EngineController {
+	return &EngineController{
+		engine:  engine,
+		log:     log,
+		metrics: metrics,
+		genesis: &genesis,
+	}
+}
+
+// State Getters
+
+func (e *EngineController) EngineSyncTarget() eth.L2BlockRef {
+	return e.syncTarget
+}
+
+func (e *EngineController) UnsafeL2Head() eth.L2BlockRef {
+	return e.unsafeHead
+}
+
+func (e *EngineController) PendingSafeL2Head() eth.L2BlockRef {
+	return e.pendingSafeHead
+}
+
+func (e *EngineController) SafeL2Head() eth.L2BlockRef {
+	return e.safeHead
+}
+
+func (e *EngineController) Finalized() eth.L2BlockRef {
+	return e.finalizedHead
+}
+
+func (e *EngineController) BuildingPayload() (eth.L2BlockRef, eth.PayloadID, bool) {
+	return e.buildingOnto, e.buildingID, e.buildingSafe
+}
+
+func (e *EngineController) IsEngineSyncing() bool {
+	return e.unsafeHead.Hash != e.syncTarget.Hash
+}
+
+// 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
+}
+
+// SetPendingSafeL2Head implements LocalEngineControl.
+func (e *EngineController) SetPendingSafeL2Head(r eth.L2BlockRef) {
+	e.metrics.RecordL2Ref("l2_pending_safe", r)
+	e.pendingSafeHead = r
+}
+
+// SetSafeHead implements LocalEngineControl.
+func (e *EngineController) SetSafeHead(r eth.L2BlockRef) {
+	e.metrics.RecordL2Ref("l2_safe", r)
+	e.safeHead = r
+}
+
+// SetUnsafeHead implements LocalEngineControl.
+func (e *EngineController) SetUnsafeHead(r eth.L2BlockRef) {
+	e.metrics.RecordL2Ref("l2_unsafe", r)
+	e.unsafeHead = r
+}
+
+// Engine Methods
+
+func (e *EngineController) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *AttributesWithParent, updateSafe bool) (errType BlockInsertionErrType, err error) {
+	if e.IsEngineSyncing() {
+		return BlockInsertTemporaryErr, fmt.Errorf("engine is in progess of p2p sync")
+	}
+	if e.buildingID != (eth.PayloadID{}) {
+		e.log.Warn("did not finish previous block building, starting new building now", "prev_onto", e.buildingOnto, "prev_payload_id", e.buildingID, "new_onto", parent)
+		// TODO(8841): maybe worth it to force-cancel the old payload ID here.
+	}
+	fc := eth.ForkchoiceState{
+		HeadBlockHash:      parent.Hash,
+		SafeBlockHash:      e.safeHead.Hash,
+		FinalizedBlockHash: e.finalizedHead.Hash,
+	}
+
+	id, errTyp, err := startPayload(ctx, e.engine, fc, attrs.attributes)
+	if err != nil {
+		return errTyp, err
+	}
+
+	e.buildingID = id
+	e.buildingSafe = updateSafe
+	e.buildingOnto = parent
+	if updateSafe {
+		e.safeAttrs = attrs
+	}
+
+	return BlockInsertOK, nil
+}
+
+func (e *EngineController) ConfirmPayload(ctx context.Context) (out *eth.ExecutionPayload, errTyp BlockInsertionErrType, err error) {
+	if e.buildingID == (eth.PayloadID{}) {
+		return nil, BlockInsertPrestateErr, fmt.Errorf("cannot complete payload building: not currently building a payload")
+	}
+	if e.buildingOnto.Hash != e.unsafeHead.Hash { // E.g. when safe-attributes consolidation fails, it will drop the existing work.
+		e.log.Warn("engine is building block that reorgs previous unsafe head", "onto", e.buildingOnto, "unsafe", e.unsafeHead)
+	}
+	fc := eth.ForkchoiceState{
+		HeadBlockHash:      common.Hash{}, // gets overridden
+		SafeBlockHash:      e.safeHead.Hash,
+		FinalizedBlockHash: e.finalizedHead.Hash,
+	}
+	// Update the safe head if the payload is built with the last attributes in the batch.
+	updateSafe := e.buildingSafe && e.safeAttrs != nil && e.safeAttrs.isLastInSpan
+	payload, errTyp, err := confirmPayload(ctx, e.log, e.engine, fc, e.buildingID, updateSafe)
+	if err != nil {
+		return nil, errTyp, fmt.Errorf("failed to complete building on top of L2 chain %s, id: %s, error (%d): %w", e.buildingOnto, e.buildingID, errTyp, err)
+	}
+	ref, err := PayloadToBlockRef(payload, e.genesis)
+	if err != nil {
+		return nil, BlockInsertPayloadErr, NewResetError(fmt.Errorf("failed to decode L2 block ref from payload: %w", err))
+	}
+
+	e.unsafeHead = ref
+	e.syncTarget = ref
+
+	e.metrics.RecordL2Ref("l2_unsafe", ref)
+	e.metrics.RecordL2Ref("l2_engineSyncTarget", ref)
+	if e.buildingSafe {
+		e.metrics.RecordL2Ref("l2_pending_safe", ref)
+		e.pendingSafeHead = ref
+		if updateSafe {
+			e.safeHead = ref
+			e.metrics.RecordL2Ref("l2_safe", ref)
+		}
+	}
+
+	e.resetBuildingState()
+	return payload, BlockInsertOK, nil
+}
+
+func (e *EngineController) CancelPayload(ctx context.Context, force bool) error {
+	if e.buildingID == (eth.PayloadID{}) { // only cancel if there is something to cancel.
+		return nil
+	}
+	// the building job gets wrapped up as soon as the payload is retrieved, there's no explicit cancel in the Engine API
+	e.log.Error("cancelling old block sealing job", "payload", e.buildingID)
+	_, err := e.engine.GetPayload(ctx, e.buildingID)
+	if err != nil {
+		e.log.Error("failed to cancel block building job", "payload", e.buildingID, "err", err)
+		if !force {
+			return err
+		}
+	}
+	e.resetBuildingState()
+	return nil
+}
+
+func (e *EngineController) resetBuildingState() {
+	e.buildingID = eth.PayloadID{}
+	e.buildingOnto = eth.L2BlockRef{}
+	e.buildingSafe = false
+	e.safeAttrs = nil
+}
+
+// Misc Setters only used by the engine queue
+
+// ResetBuildingState implements LocalEngineControl.
+func (e *EngineController) ResetBuildingState() {
+	e.resetBuildingState()
+}
+
+// ForkchoiceUpdate implements LocalEngineControl.
+func (e *EngineController) ForkchoiceUpdate(ctx context.Context, state *eth.ForkchoiceState, attr *eth.PayloadAttributes) (*eth.ForkchoiceUpdatedResult, error) {
+	return e.engine.ForkchoiceUpdate(ctx, state, attr)
+}
+
+// NewPayload implements LocalEngineControl.
+func (e *EngineController) NewPayload(ctx context.Context, payload *eth.ExecutionPayload) (*eth.PayloadStatusV1, error) {
+	return e.engine.NewPayload(ctx, payload)
+}

op-node/rollup/derive/engine_queue_test.go

@@ -258,13 +258,13 @@ func TestEngineQueue_Finalize(t *testing.T) {
 	// now say C1 was included in D and became the new safe head
 	eq.origin = refD
 	prev.origin = refD
-	eq.safeHead = refC1
+	eq.ec.SetSafeHead(refC1)
 	eq.postProcessSafeL2()
 
 	// now say D0 was included in E and became the new safe head
 	eq.origin = refE
 	prev.origin = refE
-	eq.safeHead = refD0
+	eq.ec.SetSafeHead(refD0)
 	eq.postProcessSafeL2()
 
 	// let's finalize D (current L1), from which we fully derived C1 (it was safe head), but not D0 (included in E)
@@ -275,6 +275,7 @@ func TestEngineQueue_Finalize(t *testing.T) {
 	l1F.AssertExpectations(t)
 	eng.AssertExpectations(t)
 }
+
 func TestEngineQueue_ResetWhenUnsafeOriginNotCanonical(t *testing.T) {
 	logger := testlog.Logger(t, log.LvlInfo)
 
@@ -492,14 +493,14 @@ func TestEngineQueue_ResetWhenUnsafeOriginNotCanonical(t *testing.T) {
 	// First step after reset will do a fork choice update
 	require.True(t, eq.needForkchoiceUpdate)
 	eng.ExpectForkchoiceUpdate(&eth.ForkchoiceState{
-		HeadBlockHash:      eq.unsafeHead.Hash,
-		SafeBlockHash:      eq.safeHead.Hash,
-		FinalizedBlockHash: eq.finalized.Hash,
+		HeadBlockHash:      eq.ec.UnsafeL2Head().Hash,
+		SafeBlockHash:      eq.ec.SafeL2Head().Hash,
+		FinalizedBlockHash: eq.ec.Finalized().Hash,
 	}, nil, &eth.ForkchoiceUpdatedResult{PayloadStatus: eth.PayloadStatusV1{Status: eth.ExecutionValid}}, nil)
 	err := eq.Step(context.Background())
 	require.NoError(t, err)
 
-	require.Equal(t, refF.ID(), eq.unsafeHead.L1Origin, "should have refF as unsafe head origin")
+	require.Equal(t, refF.ID(), eq.ec.UnsafeL2Head().L1Origin, "should have refF as unsafe head origin")
 
 	// L1 chain reorgs so new origin is at same slot as refF but on a different fork
 	prev.origin = eth.L1BlockRef{
@@ -823,14 +824,14 @@ func TestVerifyNewL1Origin(t *testing.T) {
 			// First step after reset will do a fork choice update
 			require.True(t, eq.needForkchoiceUpdate)
 			eng.ExpectForkchoiceUpdate(&eth.ForkchoiceState{
-				HeadBlockHash:      eq.unsafeHead.Hash,
-				SafeBlockHash:      eq.safeHead.Hash,
-				FinalizedBlockHash: eq.finalized.Hash,
+				HeadBlockHash:      eq.ec.UnsafeL2Head().Hash,
+				SafeBlockHash:      eq.ec.SafeL2Head().Hash,
+				FinalizedBlockHash: eq.ec.Finalized().Hash,
 			}, nil, &eth.ForkchoiceUpdatedResult{PayloadStatus: eth.PayloadStatusV1{Status: eth.ExecutionValid}}, nil)
 			err := eq.Step(context.Background())
 			require.NoError(t, err)
 
-			require.Equal(t, refF.ID(), eq.unsafeHead.L1Origin, "should have refF as unsafe head origin")
+			require.Equal(t, refF.ID(), eq.ec.UnsafeL2Head().L1Origin, "should have refF as unsafe head origin")
 
 			// L1 chain reorgs so new origin is at same slot as refF but on a different fork
 			prev.origin = test.newOrigin
@@ -1082,10 +1083,10 @@ func TestResetLoop(t *testing.T) {
 	prev := &fakeAttributesQueue{origin: refA, attrs: attrs, islastInSpan: true}
 
 	eq := NewEngineQueue(logger, cfg, eng, metrics.NoopMetrics, prev, l1F, &sync.Config{})
-	eq.unsafeHead = refA2
-	eq.engineSyncTarget = refA2
-	eq.safeHead = refA1
-	eq.finalized = refA0
+	eq.ec.SetUnsafeHead(refA2)
+	eq.ec.SetEngineSyncTarget(refA2)
+	eq.ec.SetSafeHead(refA1)
+	eq.ec.SetFinalizedHead(refA0)
 
 	// Queue up the safe attributes
 	require.Nil(t, eq.safeAttributes)
@@ -1180,9 +1181,9 @@ func TestEngineQueue_StepPopOlderUnsafe(t *testing.T) {
 	prev := &fakeAttributesQueue{origin: refA}
 
 	eq := NewEngineQueue(logger, cfg, eng, metrics.NoopMetrics, prev, l1F, &sync.Config{})
-	eq.unsafeHead = refA2
-	eq.safeHead = refA0
-	eq.finalized = refA0
+	eq.ec.SetUnsafeHead(refA2)
+	eq.ec.SetSafeHead(refA0)
+	eq.ec.SetFinalizedHead(refA0)
 
 	eq.AddUnsafePayload(payloadA1)
 

op-node/rollup/derive/engine_update.go

@@ -79,9 +79,9 @@ const (
 	BlockInsertPayloadErr
 )
 
-// StartPayload starts an execution payload building process in the provided Engine, with the given attributes.
+// startPayload starts an execution payload building process in the provided Engine, with the given attributes.
 // The severity of the error is distinguished to determine whether the same payload attributes may be re-attempted later.
-func StartPayload(ctx context.Context, eng Engine, fc eth.ForkchoiceState, attrs *eth.PayloadAttributes) (id eth.PayloadID, errType BlockInsertionErrType, err error) {
+func startPayload(ctx context.Context, eng ExecEngine, fc eth.ForkchoiceState, attrs *eth.PayloadAttributes) (id eth.PayloadID, errType BlockInsertionErrType, err error) {
 	fcRes, err := eng.ForkchoiceUpdate(ctx, &fc, attrs)
 	if err != nil {
 		var inputErr eth.InputError
@@ -114,10 +114,10 @@ func StartPayload(ctx context.Context, eng Engine, fc eth.ForkchoiceState, attrs
 	}
 }
 
-// ConfirmPayload ends an execution payload building process in the provided Engine, and persists the payload as the canonical head.
+// confirmPayload ends an execution payload building process in the provided Engine, and persists the payload as the canonical head.
 // If updateSafe is true, then the payload will also be recognized as safe-head at the same time.
 // The severity of the error is distinguished to determine whether the payload was valid and can become canonical.
-func ConfirmPayload(ctx context.Context, log log.Logger, eng Engine, fc eth.ForkchoiceState, id eth.PayloadID, updateSafe bool) (out *eth.ExecutionPayload, errTyp BlockInsertionErrType, err error) {
+func confirmPayload(ctx context.Context, log log.Logger, eng ExecEngine, fc eth.ForkchoiceState, id eth.PayloadID, updateSafe bool) (out *eth.ExecutionPayload, errTyp BlockInsertionErrType, err error) {
 	payload, err := eng.GetPayload(ctx, id)
 	if err != nil {
 		// even if it is an input-error (unknown payload ID), it is temporary, since we will re-attempt the full payload building, not just the retrieval of the payload.

op-node/rollup/derive/pipeline.go

@@ -56,7 +56,6 @@ type EngineQueueStage interface {
 	EngineSyncTarget() eth.L2BlockRef
 	Origin() eth.L1BlockRef
 	SystemConfig() eth.SystemConfig
-	SetUnsafeHead(head eth.L2BlockRef)
 
 	Finalize(l1Origin eth.L1BlockRef)
 	AddUnsafePayload(payload *eth.ExecutionPayload)
@@ -163,7 +162,7 @@ func (dp *DerivationPipeline) EngineSyncTarget() eth.L2BlockRef {
 	return dp.eng.EngineSyncTarget()
 }
 
-func (dp *DerivationPipeline) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *eth.PayloadAttributes, updateSafe bool) (errType BlockInsertionErrType, err error) {
+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/metered_engine.go

@@ -54,7 +54,7 @@ func (m *MeteredEngine) SafeL2Head() eth.L2BlockRef {
 	return m.inner.SafeL2Head()
 }
 
-func (m *MeteredEngine) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *eth.PayloadAttributes, updateSafe bool) (errType derive.BlockInsertionErrType, err error) {
+func (m *MeteredEngine) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *derive.AttributesWithParent, updateSafe bool) (errType derive.BlockInsertionErrType, err error) {
 	m.buildingStartTime = time.Now()
 	errType, err = m.inner.StartPayload(ctx, parent, attrs, updateSafe)
 	if err != nil {

op-node/rollup/driver/sequencer.go

@@ -96,7 +96,8 @@ func (d *Sequencer) StartBuildingBlock(ctx context.Context) error {
 		"origin", l1Origin, "origin_time", l1Origin.Time, "noTxPool", attrs.NoTxPool)
 
 	// Start a payload building process.
-	errTyp, err := d.engine.StartPayload(ctx, l2Head, attrs, false)
+	withParent := derive.NewAttributesWithParent(attrs, l2Head, false)
+	errTyp, err := d.engine.StartPayload(ctx, l2Head, withParent, false)
 	if err != nil {
 		return fmt.Errorf("failed to start building on top of L2 chain %s, error (%d): %w", l2Head, errTyp, err)
 	}

op-node/rollup/driver/sequencer_test.go

@@ -60,7 +60,7 @@ func (m *FakeEngineControl) avgTxsPerBlock() float64 {
 	return float64(m.totalTxs) / float64(m.totalBuiltBlocks)
 }
 
-func (m *FakeEngineControl) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *eth.PayloadAttributes, updateSafe bool) (errType derive.BlockInsertionErrType, err error) {
+func (m *FakeEngineControl) StartPayload(ctx context.Context, parent eth.L2BlockRef, attrs *derive.AttributesWithParent, updateSafe bool) (errType derive.BlockInsertionErrType, err error) {
 	if m.err != nil {
 		return m.errTyp, m.err
 	}
@@ -68,7 +68,7 @@ func (m *FakeEngineControl) StartPayload(ctx context.Context, parent eth.L2Block
 	_, _ = crand.Read(m.buildingID[:])
 	m.buildingOnto = parent
 	m.buildingSafe = updateSafe
-	m.buildingAttrs = attrs
+	m.buildingAttrs = attrs.Attributes()
 	m.buildingStart = m.timeNow()
 	return derive.BlockInsertOK, nil
 }