op-node/rollup: Implement Holocene invalid payload attributes handling (

#12621)

* op-node/rollup: Implement Holocene invalid payload handling

* op-node: update comment about block-processing errors

---------

Co-authored-by: protolambda <proto@protolambda.com>

op-e2e/actions/helpers/env.go

@@ -1,11 +1,14 @@
 package helpers
 
 import (
+	"math/rand"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/op-chain-ops/genesis"
+	e2ecfg "github.com/ethereum-optimism/optimism/op-e2e/config"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
@@ -16,14 +19,18 @@ type Env struct {
 	Log  log.Logger
 	Logs *testlog.CapturingHandler
 
-	SetupData *e2eutils.SetupData
+	DeployParams *e2eutils.DeployParams
+	SetupData    *e2eutils.SetupData
 
 	Miner       *L1Miner
 	Seq         *L2Sequencer
 	SeqEngine   *L2Engine
 	Verifier    *L2Verifier
 	VerifEngine *L2Engine
 	Batcher     *L2Batcher
+	Alice       *CrossLayerUser
+
+	AddressCorpora []common.Address
 }
 
 type EnvOpt struct {
@@ -49,8 +56,9 @@ const DefaultFork = rollup.Holocene
 
 // SetupEnv sets up a default action test environment. If no fork is specified, the default fork as
 // specified by the package variable [defaultFork] is used.
-func SetupEnv(t StatefulTesting, opts ...EnvOpt) (env Env) {
+func SetupEnv(t Testing, opts ...EnvOpt) (env Env) {
 	dp := e2eutils.MakeDeployParams(t, DefaultRollupTestParams())
+	env.DeployParams = dp
 
 	log, logs := testlog.CaptureLogger(t, log.LevelDebug)
 	env.Log, env.Logs = log, logs
@@ -64,16 +72,43 @@ func SetupEnv(t StatefulTesting, opts ...EnvOpt) (env Env) {
 
 	sd := e2eutils.Setup(t, dp, DefaultAlloc)
 	env.SetupData = sd
+	env.AddressCorpora = e2eutils.CollectAddresses(sd, dp)
+
 	env.Miner, env.SeqEngine, env.Seq = SetupSequencerTest(t, sd, log)
 	env.Miner.ActL1SetFeeRecipient(common.Address{'A'})
 	env.VerifEngine, env.Verifier = SetupVerifier(t, sd, log, env.Miner.L1Client(t, sd.RollupCfg), env.Miner.BlobStore(), &sync.Config{})
 	rollupSeqCl := env.Seq.RollupClient()
 	env.Batcher = NewL2Batcher(log, sd.RollupCfg, DefaultBatcherCfg(dp),
 		rollupSeqCl, env.Miner.EthClient(), env.SeqEngine.EthClient(), env.SeqEngine.EngineClient(t, sd.RollupCfg))
 
+	alice := NewCrossLayerUser(log, dp.Secrets.Alice, rand.New(rand.NewSource(0xa57b)), e2ecfg.AllocTypeStandard)
+	alice.L1.SetUserEnv(env.L1UserEnv(t))
+	alice.L2.SetUserEnv(env.L2UserEnv(t))
+	env.Alice = alice
+
 	return
 }
 
+func (env Env) L1UserEnv(t Testing) *BasicUserEnv[*L1Bindings] {
+	l1EthCl := env.Miner.EthClient()
+	return &BasicUserEnv[*L1Bindings]{
+		EthCl:          l1EthCl,
+		Signer:         types.LatestSigner(env.SetupData.L1Cfg.Config),
+		AddressCorpora: env.AddressCorpora,
+		Bindings:       NewL1Bindings(t, l1EthCl, e2ecfg.AllocTypeStandard),
+	}
+}
+
+func (env Env) L2UserEnv(t Testing) *BasicUserEnv[*L2Bindings] {
+	l2EthCl := env.SeqEngine.EthClient()
+	return &BasicUserEnv[*L2Bindings]{
+		EthCl:          l2EthCl,
+		Signer:         types.LatestSigner(env.SetupData.L2Cfg.Config),
+		AddressCorpora: env.AddressCorpora,
+		Bindings:       NewL2Bindings(t, l2EthCl, env.SeqEngine.GethClient()),
+	}
+}
+
 func (env Env) ActBatchSubmitAllAndMine(t Testing) (l1InclusionBlock *types.Block) {
 	env.Batcher.ActSubmitAll(t)
 	batchTx := env.Batcher.LastSubmitted

op-e2e/actions/helpers/l2_batcher.go

@@ -146,8 +146,8 @@ func (s *L2Batcher) Reset() {
 
 // ActL2BatchBuffer adds the next L2 block to the batch buffer.
 // If the buffer is being submitted, the buffer is wiped.
-func (s *L2Batcher) ActL2BatchBuffer(t Testing) {
-	require.NoError(t, s.Buffer(t), "failed to add block to channel")
+func (s *L2Batcher) ActL2BatchBuffer(t Testing, opts ...BlockModifier) {
+	require.NoError(t, s.Buffer(t, opts...), "failed to add block to channel")
 }
 
 type BlockModifier = func(block *types.Block)

op-e2e/actions/proofs/bad_tx_in_batch_test.go

@@ -26,14 +26,13 @@ func runBadTxInBatchTest(gt *testing.T, testCfg *helpers.TestCfg[any]) {
 	env.Alice.L2.ActCheckReceiptStatusOfLastTx(true)(t)
 
 	// Instruct the batcher to submit a faulty channel, with an invalid tx.
-	err := env.Batcher.Buffer(t, func(block *types.Block) {
+	env.Batcher.ActL2BatchBuffer(t, func(block *types.Block) {
 		// Replace the tx with one that has a bad signature.
 		txs := block.Transactions()
 		newTx, err := txs[1].WithSignature(env.Alice.L2.Signer(), make([]byte, 65))
 		txs[1] = newTx
 		require.NoError(t, err)
 	})
-	require.NoError(t, err)
 	env.Batcher.ActL2ChannelClose(t)
 	env.Batcher.ActL2BatchSubmit(t)
 

op-e2e/actions/upgrades/holocene_fork_test.go

@@ -1,13 +1,18 @@
 package upgrades
 
 import (
+	"context"
+	"math/big"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum/go-ethereum/core/types"
+
 	"github.com/ethereum-optimism/optimism/op-e2e/actions/helpers"
 	"github.com/ethereum-optimism/optimism/op-e2e/system/e2esys"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
-	"github.com/stretchr/testify/require"
 )
 
 func TestHoloceneActivationAtGenesis(gt *testing.T) {
@@ -127,3 +132,85 @@ func TestHoloceneLateActivationAndReset(gt *testing.T) {
 	requirePreHoloceneActivationLogs(e2esys.RoleSeq, 4)
 	requirePreHoloceneActivationLogs(e2esys.RoleVerif, 4)
 }
+
+func TestHoloceneInvalidPayload(gt *testing.T) {
+	t := helpers.NewDefaultTesting(gt)
+	env := helpers.SetupEnv(t, helpers.WithActiveGenesisFork(rollup.Holocene))
+	ctx := context.Background()
+
+	requireDepositOnlyLogs := func(role string, expNumLogs int) {
+		t.Helper()
+		recs := env.Logs.FindLogs(testlog.NewMessageContainsFilter("deposits-only attributes"), testlog.NewAttributesFilter("role", role))
+		require.Len(t, recs, expNumLogs)
+	}
+
+	// Start op-nodes
+	env.Seq.ActL2PipelineFull(t)
+
+	// generate and batch buffer two empty blocks
+	env.Seq.ActL2EmptyBlock(t) // 1 - genesis is 0
+	env.Batcher.ActL2BatchBuffer(t)
+	env.Seq.ActL2EmptyBlock(t) // 2
+	env.Batcher.ActL2BatchBuffer(t)
+
+	// send and include a single transaction
+	env.Alice.L2.ActResetTxOpts(t)
+	env.Alice.L2.ActSetTxToAddr(&env.DeployParams.Addresses.Bob)
+	env.Alice.L2.ActMakeTx(t)
+
+	env.Seq.ActL2StartBlock(t)
+	env.SeqEngine.ActL2IncludeTx(env.Alice.Address())(t)
+	env.Seq.ActL2EndBlock(t) // 3
+	env.Alice.L2.ActCheckReceiptStatusOfLastTx(true)(t)
+	l2Unsafe := env.Seq.L2Unsafe()
+	const invalidNum = 3
+	require.EqualValues(t, invalidNum, l2Unsafe.Number)
+	b, err := env.SeqEngine.EthClient().BlockByNumber(ctx, big.NewInt(invalidNum))
+	require.NoError(t, err)
+	require.Len(t, b.Transactions(), 2)
+
+	// buffer into the batcher, invalidating the tx via signature zeroing
+	env.Batcher.ActL2BatchBuffer(t, func(block *types.Block) {
+		// Replace the tx with one that has a bad signature.
+		txs := block.Transactions()
+		newTx, err := txs[1].WithSignature(env.Alice.L2.Signer(), make([]byte, 65))
+		require.NoError(t, err)
+		txs[1] = newTx
+	})
+
+	// generate two more empty blocks
+	env.Seq.ActL2EmptyBlock(t) // 4
+	env.Seq.ActL2EmptyBlock(t) // 5
+	require.EqualValues(t, 5, env.Seq.L2Unsafe().Number)
+
+	// submit it all
+	env.ActBatchSubmitAllAndMine(t)
+
+	// derive chain on sequencer
+	env.Seq.ActL1HeadSignal(t)
+	env.Seq.ActL2PipelineFull(t)
+
+	// TODO(12695): need to properly update safe after completed L1 block derivation
+	l2Safe := env.Seq.L2PendingSafe()
+	require.EqualValues(t, invalidNum, l2Safe.Number)
+	require.NotEqual(t, l2Safe.Hash, l2Unsafe.Hash, // old L2Unsafe above
+		"block-3 should have been replaced by deposit-only version")
+	requireDepositOnlyLogs(e2esys.RoleSeq, 2)
+	require.Equal(t, l2Safe, env.Seq.L2Unsafe(), "unsafe chain should have reorg'd")
+	b, err = env.SeqEngine.EthClient().BlockByNumber(ctx, big.NewInt(invalidNum))
+	require.NoError(t, err)
+	require.Len(t, b.Transactions(), 1)
+
+	// test that building on top of reorg'd chain and deriving further works
+
+	env.Seq.ActL2EmptyBlock(t) // 4
+	env.Seq.ActL2EmptyBlock(t) // 5
+	l2Unsafe = env.Seq.L2Unsafe()
+	require.EqualValues(t, 5, l2Unsafe.Number)
+
+	env.Batcher.Reset() // need to reset batcher to become aware of reorg
+	env.ActBatchSubmitAllAndMine(t)
+	env.Seq.ActL1HeadSignal(t)
+	env.Seq.ActL2PipelineFull(t)
+	require.Equal(t, l2Unsafe, env.Seq.L2Safe())
+}

op-node/rollup/attributes/attributes.go

@@ -62,6 +62,7 @@ func (eq *AttributesHandler) OnEvent(ev event.Event) bool {
 		eq.onPendingSafeUpdate(x)
 	case derive.DerivedAttributesEvent:
 		eq.attributes = x.Attributes
+		eq.sentAttributes = false
 		eq.emitter.Emit(derive.ConfirmReceivedAttributesEvent{})
 		// to make sure we have a pre-state signal to process the attributes from
 		eq.emitter.Emit(engine.PendingSafeRequestEvent{})
@@ -71,7 +72,7 @@ func (eq *AttributesHandler) OnEvent(ev event.Event) bool {
 	case rollup.EngineTemporaryErrorEvent:
 		eq.sentAttributes = false
 	case engine.InvalidPayloadAttributesEvent:
-		if x.Attributes.DerivedFrom == (eth.L1BlockRef{}) {
+		if !x.Attributes.IsDerived() {
 			return true // from sequencing
 		}
 		eq.sentAttributes = false

op-node/rollup/derive/attributes_queue.go

@@ -2,6 +2,7 @@ package derive
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -35,17 +36,32 @@ type AttributesWithParent struct {
 	DerivedFrom eth.L1BlockRef
 }
 
+// WithDepositsOnly return a shallow clone with all non-Deposit transactions
+// stripped from the transactions of its attributes. The order is preserved.
+func (a *AttributesWithParent) WithDepositsOnly() *AttributesWithParent {
+	clone := *a
+	clone.Attributes = clone.Attributes.WithDepositsOnly()
+	return &clone
+}
+
+func (a *AttributesWithParent) IsDerived() bool {
+	return a.DerivedFrom != (eth.L1BlockRef{})
+}
+
 type AttributesQueue struct {
-	log          log.Logger
-	config       *rollup.Config
-	builder      AttributesBuilder
-	prev         SingularBatchProvider
+	log     log.Logger
+	config  *rollup.Config
+	builder AttributesBuilder
+	prev    SingularBatchProvider
+
 	batch        *SingularBatch
 	isLastInSpan bool
+	lastAttribs  *AttributesWithParent
 }
 
 type SingularBatchProvider interface {
 	ResettableStage
+	ChannelFlusher
 	Origin() eth.L1BlockRef
 	NextBatch(context.Context, eth.L2BlockRef) (*SingularBatch, bool, error)
 }
@@ -85,11 +101,11 @@ func (aq *AttributesQueue) NextAttributes(ctx context.Context, parent eth.L2Bloc
 			IsLastInSpan: aq.isLastInSpan,
 			DerivedFrom:  aq.Origin(),
 		}
+		aq.lastAttribs = &attr
 		aq.batch = nil
 		aq.isLastInSpan = false
 		return &attr, nil
 	}
-
 }
 
 // createNextAttributes transforms a batch into a payload attributes. This sets `NoTxPool` and appends the batched transactions
@@ -120,8 +136,35 @@ func (aq *AttributesQueue) createNextAttributes(ctx context.Context, batch *Sing
 	return attrs, nil
 }
 
-func (aq *AttributesQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.SystemConfig) error {
+func (aq *AttributesQueue) reset() {
 	aq.batch = nil
 	aq.isLastInSpan = false // overwritten later, but set for consistency
+	aq.lastAttribs = nil
+}
+
+func (aq *AttributesQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.SystemConfig) error {
+	aq.reset()
 	return io.EOF
 }
+
+func (aq *AttributesQueue) DepositsOnlyAttributes(parent eth.BlockID, derivedFrom eth.L1BlockRef) (*AttributesWithParent, error) {
+	// Sanity checks - these cannot happen with correct deriver implementations.
+	if aq.batch != nil {
+		return nil, fmt.Errorf("unexpected buffered batch, parent hash: %s, epoch: %s", aq.batch.ParentHash, aq.batch.Epoch())
+	} else if aq.lastAttribs == nil {
+		return nil, errors.New("no attributes generated yet")
+	} else if derivedFrom != aq.lastAttribs.DerivedFrom {
+		return nil, fmt.Errorf(
+			"unexpected derivation origin, last_origin: %s, invalid_origin: %s",
+			aq.lastAttribs.DerivedFrom, derivedFrom)
+	} else if parent != aq.lastAttribs.Parent.ID() {
+		return nil, fmt.Errorf(
+			"unexpected parent: last_parent: %s, invalid_parent: %s",
+			aq.lastAttribs.Parent.ID(), parent)
+	}
+
+	aq.prev.FlushChannel() // flush all channel data in previous stages
+	attrs := aq.lastAttribs.WithDepositsOnly()
+	aq.lastAttribs = attrs
+	return attrs, nil
+}

op-node/rollup/derive/batch_queue.go

@@ -27,10 +27,6 @@ import (
 // It is internally responsible for making sure that batches with L1 inclusions block outside it's
 // working range are not considered or pruned.
 
-type ChannelFlusher interface {
-	FlushChannel()
-}
-
 type NextBatchProvider interface {
 	ChannelFlusher
 	Origin() eth.L1BlockRef
@@ -271,6 +267,12 @@ func (bq *BatchQueue) Reset(_ context.Context, base eth.L1BlockRef, _ eth.System
 	return io.EOF
 }
 
+func (bq *BatchQueue) FlushChannel() {
+	// We need to implement the ChannelFlusher interface with the BatchQueue but it's never called
+	// of which the BatchMux takes care.
+	panic("BatchQueue: invalid FlushChannel call")
+}
+
 func (bq *BatchQueue) AddBatch(ctx context.Context, batch Batch, parent eth.L2BlockRef) {
 	if len(bq.l1Blocks) == 0 {
 		panic(fmt.Errorf("cannot add batch with timestamp %d, no origin was prepared", batch.GetTimestamp()))

op-node/rollup/derive/channel_assembler.go

@@ -51,6 +51,10 @@ func (ca *ChannelAssembler) Reset(context.Context, eth.L1BlockRef, eth.SystemCon
 	return io.EOF
 }
 
+func (ca *ChannelAssembler) FlushChannel() {
+	ca.resetChannel()
+}
+
 func (ca *ChannelAssembler) resetChannel() {
 	ca.channel = nil
 }

op-node/rollup/derive/channel_bank.go

@@ -203,6 +203,12 @@ func (cb *ChannelBank) Reset(ctx context.Context, base eth.L1BlockRef, _ eth.Sys
 	return io.EOF
 }
 
+func (bq *ChannelBank) FlushChannel() {
+	// We need to implement the ChannelFlusher interface with the ChannelBank but it's never called
+	// of which the ChannelMux takes care.
+	panic("ChannelBank: invalid FlushChannel call")
+}
+
 type L1BlockRefByHashFetcher interface {
 	L1BlockRefByHash(context.Context, common.Hash) (eth.L1BlockRef, error)
 }

op-node/rollup/derive/channel_in_reader.go

@@ -32,6 +32,7 @@ var (
 
 type RawChannelProvider interface {
 	ResettableStage
+	ChannelFlusher
 	Origin() eth.L1BlockRef
 	NextRawChannel(ctx context.Context) ([]byte, error)
 }
@@ -134,5 +135,5 @@ func (cr *ChannelInReader) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.Sy
 
 func (cr *ChannelInReader) FlushChannel() {
 	cr.nextBatchFn = nil
-	// TODO(12157): cr.prev.FlushChannel() - when we do wiring with ChannelStage
+	cr.prev.FlushChannel()
 }