Merge pull request #5421 from ethereum-optimism/jg/fix_reset_loop

op-node: Fix reset loop

op-node/rollup/derive/engine_queue.go

@@ -17,6 +17,11 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 )
 
+type attributesWithParent struct {
+	attributes *eth.PayloadAttributes
+	parent     eth.L2BlockRef
+}
+
 type NextAttributesProvider interface {
 	Origin() eth.L1BlockRef
 	NextAttributes(context.Context, eth.L2BlockRef) (*eth.PayloadAttributes, error)
@@ -114,9 +119,8 @@ type EngineQueue struct {
 	triedFinalizeAt eth.L1BlockRef
 
 	// The queued-up attributes
-	safeAttributesParent eth.L2BlockRef
-	safeAttributes       *eth.PayloadAttributes
-	unsafePayloads       *PayloadsQueue // queue of unsafe payloads, ordered by ascending block number, may have gaps and duplicates
+	safeAttributes *attributesWithParent
+	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
@@ -241,9 +245,11 @@ func (eq *EngineQueue) Step(ctx context.Context) error {
 	} else if err != nil {
 		return err
 	} else {
-		eq.safeAttributes = next
-		eq.safeAttributesParent = eq.safeHead
-		eq.log.Debug("Adding next safe attributes", "safe_head", eq.safeHead, "next", eq.safeAttributes)
+		eq.safeAttributes = &attributesWithParent{
+			attributes: next,
+			parent:     eq.safeHead,
+		}
+		eq.log.Debug("Adding next safe attributes", "safe_head", eq.safeHead, "next", next)
 		return NotEnoughData
 	}
 
@@ -482,15 +488,19 @@ func (eq *EngineQueue) tryNextSafeAttributes(ctx context.Context) error {
 		return nil
 	}
 	// validate the safe attributes before processing them. The engine may have completed processing them through other means.
-	if eq.safeHead != eq.safeAttributesParent {
-		if eq.safeHead.ParentHash != eq.safeAttributesParent.Hash {
-			return NewResetError(fmt.Errorf("safe head changed to %s with parent %s, conflicting with queued safe attributes on top of %s",
-				eq.safeHead, eq.safeHead.ParentID(), eq.safeAttributesParent))
+	if eq.safeHead != eq.safeAttributes.parent {
+		// Previously the attribute's parent was the safe head. If the safe head advances so safe head's parent is the same as the
+		// attribute's parent then we need to cancel the attributes.
+		if eq.safeHead.ParentHash == eq.safeAttributes.parent.Hash {
+			eq.log.Warn("queued safe attributes are stale, safehead progressed",
+				"safe_head", eq.safeHead, "safe_head_parent", eq.safeHead.ParentID(), "attributes_parent", eq.safeAttributes.parent)
+			eq.safeAttributes = nil
+			return nil
 		}
-		eq.log.Warn("queued safe attributes are stale, safe-head progressed",
-			"safe_head", eq.safeHead, "safe_head_parent", eq.safeHead.ParentID(), "attributes_parent", eq.safeAttributesParent)
-		eq.safeAttributes = nil
-		return nil
+		// If something other than a simple advance occurred, perform a full reset
+		return NewResetError(fmt.Errorf("safe head changed to %s with parent %s, conflicting with queued safe attributes on top of %s",
+			eq.safeHead, eq.safeHead.ParentID(), eq.safeAttributes.parent))
+
 	}
 	if eq.safeHead.Number < eq.unsafeHead.Number {
 		return eq.consolidateNextSafeAttributes(ctx)
@@ -520,7 +530,7 @@ func (eq *EngineQueue) consolidateNextSafeAttributes(ctx context.Context) error
 		}
 		return NewTemporaryError(fmt.Errorf("failed to get existing unsafe payload to compare against derived attributes from L1: %w", err))
 	}
-	if err := AttributesMatchBlock(eq.safeAttributes, eq.safeHead.Hash, payload, eq.log); err != nil {
+	if err := AttributesMatchBlock(eq.safeAttributes.attributes, eq.safeHead.Hash, payload, eq.log); err != nil {
 		eq.log.Warn("L2 reorg: existing unsafe block does not match derived attributes from L1", "err", err, "unsafe", eq.unsafeHead, "safe", eq.safeHead)
 		// geth cannot wind back a chain without reorging to a new, previously non-canonical, block
 		return eq.forceNextSafeAttributes(ctx)
@@ -545,7 +555,7 @@ func (eq *EngineQueue) forceNextSafeAttributes(ctx context.Context) error {
 	if eq.safeAttributes == nil {
 		return nil
 	}
-	attrs := eq.safeAttributes
+	attrs := eq.safeAttributes.attributes
 	errType, err := eq.StartPayload(ctx, eq.safeHead, attrs, true)
 	if err == nil {
 		_, errType, err = eq.ConfirmPayload(ctx)
@@ -716,6 +726,7 @@ func (eq *EngineQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.System
 	eq.log.Debug("Reset engine queue", "safeHead", safe, "unsafe", unsafe, "safe_timestamp", safe.Time, "unsafe_timestamp", unsafe.Time, "l1Origin", l1Origin)
 	eq.unsafeHead = unsafe
 	eq.safeHead = safe
+	eq.safeAttributes = nil
 	eq.finalized = finalized
 	eq.resetBuildingState()
 	eq.needForkchoiceUpdate = true

op-node/rollup/derive/engine_queue_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/testlog"
 	"github.com/ethereum-optimism/optimism/op-node/testutils"
@@ -1007,3 +1008,102 @@ func TestBlockBuildingRace(t *testing.T) {
 	l1F.AssertExpectations(t)
 	eng.AssertExpectations(t)
 }
+
+func TestResetLoop(t *testing.T) {
+	logger := testlog.Logger(t, log.LvlInfo)
+	eng := &testutils.MockEngine{}
+	l1F := &testutils.MockL1Source{}
+
+	rng := rand.New(rand.NewSource(1234))
+
+	refA := testutils.RandomBlockRef(rng)
+	refA0 := eth.L2BlockRef{
+		Hash:           testutils.RandomHash(rng),
+		Number:         0,
+		ParentHash:     common.Hash{},
+		Time:           refA.Time,
+		L1Origin:       refA.ID(),
+		SequenceNumber: 0,
+	}
+	gasLimit := eth.Uint64Quantity(20_000_000)
+	cfg := &rollup.Config{
+		Genesis: rollup.Genesis{
+			L1:     refA.ID(),
+			L2:     refA0.ID(),
+			L2Time: refA0.Time,
+			SystemConfig: eth.SystemConfig{
+				BatcherAddr: common.Address{42},
+				Overhead:    [32]byte{123},
+				Scalar:      [32]byte{42},
+				GasLimit:    20_000_000,
+			},
+		},
+		BlockTime:     1,
+		SeqWindowSize: 2,
+	}
+	refA1 := eth.L2BlockRef{
+		Hash:           testutils.RandomHash(rng),
+		Number:         refA0.Number + 1,
+		ParentHash:     refA0.Hash,
+		Time:           refA0.Time + cfg.BlockTime,
+		L1Origin:       refA.ID(),
+		SequenceNumber: 1,
+	}
+	refA2 := eth.L2BlockRef{
+		Hash:           testutils.RandomHash(rng),
+		Number:         refA1.Number + 1,
+		ParentHash:     refA1.Hash,
+		Time:           refA1.Time + cfg.BlockTime,
+		L1Origin:       refA.ID(),
+		SequenceNumber: 2,
+	}
+
+	attrs := &eth.PayloadAttributes{
+		Timestamp:             eth.Uint64Quantity(refA2.Time),
+		PrevRandao:            eth.Bytes32{},
+		SuggestedFeeRecipient: common.Address{},
+		Transactions:          nil,
+		NoTxPool:              false,
+		GasLimit:              &gasLimit,
+	}
+
+	eng.ExpectL2BlockRefByLabel(eth.Finalized, refA0, nil)
+	eng.ExpectL2BlockRefByLabel(eth.Safe, refA1, nil)
+	eng.ExpectL2BlockRefByLabel(eth.Unsafe, refA2, nil)
+	eng.ExpectL2BlockRefByHash(refA1.Hash, refA1, nil)
+	eng.ExpectL2BlockRefByHash(refA0.Hash, refA0, nil)
+	eng.ExpectSystemConfigByL2Hash(refA0.Hash, cfg.Genesis.SystemConfig, nil)
+	l1F.ExpectL1BlockRefByNumber(refA.Number, refA, nil)
+	l1F.ExpectL1BlockRefByHash(refA.Hash, refA, nil)
+	l1F.ExpectL1BlockRefByHash(refA.Hash, refA, nil)
+
+	prev := &fakeAttributesQueue{origin: refA, attrs: attrs}
+
+	eq := NewEngineQueue(logger, cfg, eng, metrics.NoopMetrics, prev, l1F)
+	eq.unsafeHead = refA2
+	eq.safeHead = refA1
+	eq.finalized = refA0
+
+	// Qeueue up the safe attributes
+	require.Nil(t, eq.safeAttributes)
+	require.ErrorIs(t, eq.Step(context.Background()), NotEnoughData)
+	require.NotNil(t, eq.safeAttributes)
+
+	// Peform the reset
+	require.ErrorIs(t, eq.Reset(context.Background(), eth.L1BlockRef{}, eth.SystemConfig{}), io.EOF)
+
+	// Expect a FCU after the reset
+	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()), "clean forkchoice state after reset")
+
+	// Crux of the test. Should be in a valid state after the reset.
+	require.ErrorIs(t, eq.Step(context.Background()), NotEnoughData, "Should be able to step after a reset")
+
+	l1F.AssertExpectations(t)
+	eng.AssertExpectations(t)
+}

op-node/testutils/mock_eth_client.go

@@ -83,8 +83,8 @@ func (m *MockEthClient) PayloadByNumber(ctx context.Context, n uint64) (*eth.Exe
 	return out[0].(*eth.ExecutionPayload), *out[1].(*error)
 }
 
-func (m *MockEthClient) ExpectPayloadByNumber(hash common.Hash, payload *eth.ExecutionPayload, err error) {
-	m.Mock.On("PayloadByNumber", hash).Once().Return(payload, &err)
+func (m *MockEthClient) ExpectPayloadByNumber(n uint64, payload *eth.ExecutionPayload, err error) {
+	m.Mock.On("PayloadByNumber", n).Once().Return(payload, &err)
 }
 
 func (m *MockEthClient) PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*eth.ExecutionPayload, error) {